Publish events for agent personas and ACLs

core/events/message.py

@@ -12,6 +12,7 @@
 
 if TYPE_CHECKING:
     from core.schemas import (
+        agent_persona,
         dfiq,
         entity,
         graph,
@@ -66,7 +67,9 @@ def yeti_object_discriminator(v):
         Annotated["tag.Tag", PydanticTag("tag")],
         Annotated["template.Template", PydanticTag("template")],
         Annotated["graph.Relationship", PydanticTag("relationship")],
+        Annotated["graph.RoleRelationship", PydanticTag("acl")],
         Annotated["rbac.Group", PydanticTag("rbacgroup")],
+        Annotated["agent_persona.AgentPersona", PydanticTag("agent_persona")],
     ],
     Field(discriminator=Discriminator(yeti_object_discriminator)),
 ]

core/schemas/agent_persona.py

@@ -1,7 +1,7 @@
 import datetime
 from typing import ClassVar, Literal
 
-from pydantic import ConfigDict, Field, field_validator
+from pydantic import ConfigDict, Field, computed_field, field_validator
 
 from core import database_arango
 from core.helpers import now
@@ -40,6 +40,11 @@ class AgentPersona(YetiModel, YetiAclModel, database_arango.ArangoYetiConnector)
     created: datetime.datetime = Field(default_factory=now)
     modified: datetime.datetime = Field(default_factory=now)
 
+    @computed_field(return_type=Literal["agent_persona"])
+    @property
+    def root_type(self):
+        return self._root_type
+
     @field_validator("instruction")
     @classmethod
     def instruction_is_substantial(cls, value: str) -> str:

tests/core_tests/events.py

@@ -8,7 +8,7 @@
 from core import database_arango
 from core.config.config import yeti_config
 from core.events import message, producer
-from core.schemas import observable
+from core.schemas import agent_persona, observable
 
 
 class EventsTest(unittest.TestCase):
@@ -133,3 +133,114 @@ def test_queue_memory_limit(self) -> None:
         body = json.loads(base64.b64decode(body_payload))
         event = message.EventMessage(**json.loads(body))
         self.assertEqual(event.event.yeti_object.value, f"test{i}.com")
+
+
+class EventUnionCoverageTest(unittest.TestCase):
+    """Guards the two ways an object can fail to become an event.
+
+    Both are silent: publishing is wrapped in a try/except that logs, so the
+    write still lands and only the log says anything. AgentPersona shipped
+    broken on both counts without any test noticing.
+    """
+
+    # Written to, but deliberately never published: save() and delete() skip
+    # these by collection name. An event per audit-log line would be circular.
+    NEVER_PUBLISHED = {"auditlog", "timeline"}
+    # Publishes a LinkEvent, which names its objects rather than discriminating
+    # on them.
+    LINK_COLLECTION = "links"
+
+    def publishing_classes(self):
+        import core.schemas  # noqa: F401  (registers every schema type)
+
+        def descendants(cls):
+            for sub in cls.__subclasses__():
+                yield sub
+                yield from descendants(sub)
+
+        seen = {}
+        for cls in descendants(database_arango.ArangoYetiConnector):
+            collection = getattr(cls, "_collection_name", None)
+            if collection in self.NEVER_PUBLISHED or collection is None:
+                continue
+            if collection == self.LINK_COLLECTION:
+                continue
+            root_type = cls.__private_attributes__.get("_root_type")
+            if root_type is not None:
+                seen[cls] = root_type.default
+        return seen
+
+    def test_every_published_object_exposes_root_type(self):
+        """The discriminator reads `root_type` off the instance. A class with
+        only the private `_root_type` resolves to no tag at all."""
+        for cls in self.publishing_classes():
+            self.assertIn(
+                "root_type",
+                cls.model_fields | cls.model_computed_fields,
+                f"{cls.__name__} does not expose root_type",
+            )
+
+    def test_every_published_root_type_has_a_union_member(self):
+        """A root_type absent from YetiObjectTypes raises union_tag_not_found
+        on every save and delete of that object."""
+        import typing
+
+        union, _ = typing.get_args(message.YetiObjectTypes)
+        tags = set()
+        for member in typing.get_args(union):
+            for meta in getattr(member, "__metadata__", ()):
+                tag = getattr(meta, "tag", None)
+                if tag:
+                    tags.add(tag)
+
+        missing = {
+            root_type
+            for root_type in self.publishing_classes().values()
+            if root_type not in tags
+        }
+        self.assertEqual(missing, set(), f"root types with no union member: {missing}")
+
+
+class AgentPersonaEventsTest(unittest.TestCase):
+    def setUp(self) -> None:
+        database_arango.db.connect(database="yeti_test")
+        database_arango.db.clear()
+        self.redis_client = redis.from_url(
+            f"redis://{yeti_config.get('redis', 'host')}/"
+        )
+        self.redis_client.delete("events")
+
+    def tearDown(self) -> None:
+        database_arango.db.clear()
+        self.redis_client.delete("events")
+
+    def persona(self):
+        return agent_persona.AgentPersona(
+            name="Default", instruction="Be helpful, and be brief about it."
+        )
+
+    def test_saving_a_persona_publishes_an_event(self) -> None:
+        saved = self.persona().save()
+
+        self.assertEqual(self.redis_client.llen("events"), 1)
+        body = json.loads(
+            base64.b64decode(json.loads(self.redis_client.lpop("events"))["body"])
+        )
+        event = message.EventMessage(**json.loads(body))
+        self.assertEqual(event.event.type, message.EventType.new)
+        self.assertEqual(event.event.yeti_object.id, saved.id)
+        self.assertEqual(event.event.yeti_object.name, "Default")
+
+    def test_deleting_a_persona_publishes_an_event(self) -> None:
+        saved = self.persona().save()
+        self.redis_client.delete("events")
+
+        saved.delete()
+
+        self.assertEqual(self.redis_client.llen("events"), 1)
+        body = json.loads(
+            base64.b64decode(json.loads(self.redis_client.lpop("events"))["body"])
+        )
+        event = message.EventMessage(**json.loads(body))
+        self.assertEqual(event.event.type, message.EventType.delete)
+        self.assertEqual(event.event.yeti_object.name, "Default")