fix(timeline): use strict agent attributes schema

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/services/timeline/contracts.py

@@ -49,6 +49,18 @@ class AgentAssertion(BaseModel):
     evidence_ids: list[str] = Field(min_length=1)
 
 
+class AgentAttribute(BaseModel):
+    """A strict-schema-safe episode attribute.
+
+    Codex structured outputs cannot represent an object with arbitrary keys, so the
+    agent emits key/value entries and Chronicle materializes them as an object when
+    publishing the episode.
+    """
+
+    key: str = Field(min_length=1, max_length=80)
+    value: str = Field(max_length=500)
+
+
 class AgentEpisode(BaseModel):
     kind: str = Field(min_length=1, max_length=80)
     title: str = Field(min_length=1, max_length=160)
@@ -59,7 +71,7 @@ class AgentEpisode(BaseModel):
     activity_mode: Literal["foreground", "background", "ambient", "idle"]
     confidence: float = Field(ge=0, le=1)
     entities: list[str] = Field(default_factory=list)
-    attributes: dict[str, Any] = Field(default_factory=dict)
+    attributes: list[AgentAttribute] = Field(default_factory=list)
     assertions: list[AgentAssertion] = Field(default_factory=list)
     evidence_ids: list[str] = Field(min_length=1)
     related_conversation_ids: list[str] = Field(default_factory=list)

backends/advanced/src/advanced_omi_backend/services/timeline/discovery.py

@@ -183,7 +183,7 @@ async def _publish(
                 confidence=episode.confidence,
                 activity_mode=episode.activity_mode,
                 entities=episode.entities,
-                attributes=episode.attributes,
+                attributes={item.key: item.value for item in episode.attributes},
                 assertions=[
                     TimelineAssertion(**assertion.model_dump())
                     for assertion in episode.assertions

backends/advanced/src/advanced_omi_backend/services/timeline/prompt.py

@@ -31,7 +31,18 @@
                     },
                     "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                     "entities": {"type": "array", "items": {"type": "string"}},
-                    "attributes": {"type": "object"},
+                    "attributes": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "additionalProperties": False,
+                            "properties": {
+                                "key": {"type": "string"},
+                                "value": {"type": "string"},
+                            },
+                            "required": ["key", "value"],
+                        },
+                    },
                     "assertions": {
                         "type": "array",
                         "items": {
@@ -131,5 +142,6 @@ def build_prompt(output_path: str) -> str:
 - Every important boundary/assertion must cite supplied evidence IDs.
 - Never invent an evidence ID. Include every window ID exactly once in covered_window_ids.
 - Salience is display value, not confidence.
+- Express optional episode metadata as short string key/value entries in `attributes`.
 - Prefer a few coherent episodes over arbitrary periodic fragments.
 """

backends/advanced/tests/test_timeline_executor.py

@@ -4,6 +4,7 @@
 
 from advanced_omi_backend.services.timeline.contracts import (
     AgentAssertion,
+    AgentAttribute,
     AgentEpisode,
     TimelineAgentResult,
     TimelineCoverageWindow,
@@ -57,6 +58,7 @@ def _result(evidence_id: str = "observation:one") -> TimelineAgentResult:
                 salience="routine",
                 activity_mode="background",
                 confidence=0.9,
+                attributes=[AgentAttribute(key="media", value="Terminator")],
                 evidence_ids=[evidence_id],
                 assertions=[
                     AgentAssertion(

backends/advanced/tests/test_timeline_prompt.py

@@ -0,0 +1,16 @@
+from advanced_omi_backend.services.timeline.prompt import OUTPUT_SCHEMA
+
+
+def _assert_strict_objects(schema: object) -> None:
+    if isinstance(schema, dict):
+        if schema.get("type") == "object":
+            assert schema.get("additionalProperties") is False
+        for value in schema.values():
+            _assert_strict_objects(value)
+    elif isinstance(schema, list):
+        for value in schema:
+            _assert_strict_objects(value)
+
+
+def test_timeline_output_schema_has_no_open_objects():
+    _assert_strict_objects(OUTPUT_SCHEMA)