Vendor generated SourceOS interaction Python types

src/agent_term/contracts/sourceos/generated/sourceos_interaction_event.py

@@ -0,0 +1,107 @@
+# Generated from schemas/SourceOSInteractionEvent.json.
+# Do not edit by hand. Source: SourceOS-Linux/sourceos-spec generated/python/sourceos_interaction_event.py
+# Pinned sourceos-spec commit: c7f8c2d9e42a56e1127c2f9b85649cbea0f0a9fa
+
+from __future__ import annotations
+
+from typing import Any, Literal, NotRequired, TypedDict
+
+
+SOURCEOS_INTERACTION_EVENT_REQUIRED = ['interactionEventId', 'type', 'specVersion', 'eventClass', 'occurredAt', 'surface', 'mode', 'session', 'actor', 'payloadMode', 'governanceTrace']
+
+SourceOSInteractionEventClass = Literal['interaction.session_started', 'interaction.message_posted', 'interaction.task_submitted', 'interaction.task_stream_delta', 'interaction.task_completed', 'interaction.task_failed', 'interaction.governance_trace', 'interaction.steering_intent', 'interaction.memory_scope_bound', 'interaction.context_pack_bound', 'interaction.policy_decision', 'interaction.approval_requested', 'interaction.approval_recorded', 'interaction.redacted']
+SourceOSInteractionSurfaceKind = Literal['noetica', 'agent-term', 'matrix', 'prophet-workspace', 'superconscious', 'agentplane', 'api', 'other']
+SourceOSInteractionMode = Literal['standalone', 'sourceos', 'dry-run', 'replay']
+SourceOSInteractionActorKind = Literal['human', 'agent', 'service', 'bot', 'system']
+SourceOSInteractionParticipantRole = Literal['user', 'assistant', 'operator', 'agent', 'provider', 'tool', 'observer']
+SourceOSInteractionTaskStatus = Literal['submitted', 'streaming', 'success', 'failure', 'blocked', 'unavailable', 'not_configured']
+SourceOSSteeringKind = Literal['none', 'neuronpedia_feature', 'local_sae', 'sourceos_local', 'other']
+SourceOSSteeringStatus = Literal['requested', 'applied', 'noop', 'not_configured', 'blocked']
+SourceOSInteractionPayloadMode = Literal['metadata-only', 'summary', 'ref-only', 'inline-bounded', 'redacted']
+
+
+class SourceOSInteractionSurface(TypedDict):
+    surfaceKind: SourceOSInteractionSurfaceKind
+    sourcePlane: str
+    clientRef: NotRequired[str | None]
+
+
+class SourceOSInteractionSession(TypedDict):
+    sessionId: str
+    conversationRef: NotRequired[str | None]
+    roomRef: NotRequired[str | None]
+    threadRef: NotRequired[str | None]
+    workroomRef: NotRequired[str | None]
+    topicRef: NotRequired[str | None]
+    opsHistoryEventRef: NotRequired[str | None]
+
+
+class SourceOSInteractionActor(TypedDict):
+    actorRef: str
+    actorKind: SourceOSInteractionActorKind
+    agentRegistryRef: NotRequired[str | None]
+    onBehalfOfRef: NotRequired[str | None]
+
+
+class SourceOSInteractionParticipant(TypedDict):
+    role: SourceOSInteractionParticipantRole
+    participantRef: str
+    agentRegistryRef: NotRequired[str | None]
+
+
+class SourceOSInteractionTask(TypedDict, total=False):
+    taskRef: str | None
+    status: SourceOSInteractionTaskStatus
+    modelHint: str | None
+    modelRouted: str | None
+    provider: str | None
+    latencyMs: int | None
+
+
+class SourceOSSteeringIntent(TypedDict, total=False):
+    steeringKind: SourceOSSteeringKind
+    featureRef: str | None
+    strength: float | None
+    status: SourceOSSteeringStatus
+
+
+class SourceOSGovernanceTrace(TypedDict):
+    policyAdmitted: bool
+    memoryWritten: bool
+    policyRef: NotRequired[str | None]
+    policyDecisionRefs: NotRequired[list[str]]
+    grantRefs: NotRequired[list[str]]
+    memoryScopeRef: NotRequired[str | None]
+    contextPackRefs: NotRequired[list[str]]
+    requestHash: NotRequired[str | None]
+    evidenceHash: NotRequired[str | None]
+    providerRouteEvidenceRef: NotRequired[str | None]
+    agentPlaneRunRef: NotRequired[str | None]
+    evidenceRefs: NotRequired[list[str]]
+    replayRef: NotRequired[str | None]
+
+
+class SourceOSInteractionIntegrity(TypedDict, total=False):
+    eventHash: str | None
+    signature: str | None
+
+
+class SourceOSInteractionEvent(TypedDict):
+    interactionEventId: str
+    type: Literal["SourceOSInteractionEvent"]
+    specVersion: str
+    eventClass: SourceOSInteractionEventClass
+    occurredAt: str
+    surface: SourceOSInteractionSurface
+    mode: SourceOSInteractionMode
+    session: SourceOSInteractionSession
+    actor: SourceOSInteractionActor
+    payloadMode: SourceOSInteractionPayloadMode
+    governanceTrace: SourceOSGovernanceTrace
+    participants: NotRequired[list[SourceOSInteractionParticipant]]
+    task: NotRequired[SourceOSInteractionTask | None]
+    steeringIntent: NotRequired[SourceOSSteeringIntent | None]
+    payload: NotRequired[dict[str, Any] | None]
+    sourceEventRefs: NotRequired[list[str]]
+    redactionRefs: NotRequired[list[str]]
+    integrity: NotRequired[SourceOSInteractionIntegrity | None]

src/agent_term/interaction.py

@@ -9,46 +9,37 @@
 
 import json
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
+from agent_term.contracts.sourceos.generated.sourceos_interaction_event import (
+    SOURCEOS_INTERACTION_EVENT_REQUIRED,
+    SourceOSInteractionEvent,
+)
 from agent_term.events import AgentTermEvent
 
 JsonObject = dict[str, Any]
 
-REQUIRED_TOP_LEVEL = {
-    "interactionEventId",
-    "type",
-    "specVersion",
-    "eventClass",
-    "occurredAt",
-    "surface",
-    "mode",
-    "session",
-    "actor",
-    "payloadMode",
-    "governanceTrace",
-}
-
+REQUIRED_TOP_LEVEL = set(SOURCEOS_INTERACTION_EVENT_REQUIRED)
 REQUIRED_GOVERNANCE = {"policyAdmitted", "memoryWritten"}
 
 
-def load_interaction_event(path: Path | str) -> JsonObject:
+def load_interaction_event(path: Path | str) -> SourceOSInteractionEvent:
     """Load a SourceOSInteractionEvent JSON object from disk."""
 
     with Path(path).open("r", encoding="utf-8") as handle:
         value = json.load(handle)
 
     if not isinstance(value, dict):
         raise ValueError("SourceOSInteractionEvent payload must be a JSON object")
-    return value
+    return cast(SourceOSInteractionEvent, value)
 
 
-def validate_interaction_event(event: JsonObject) -> list[str]:
+def validate_interaction_event(event: SourceOSInteractionEvent | JsonObject) -> list[str]:
     """Return structural validation errors for AgentTerm's render/ingest boundary.
 
-    This is intentionally a focused local check rather than a full JSON Schema validator.
-    Full schema validation belongs to sourceos-spec CI. AgentTerm needs enough validation
-    to fail closed before rendering or recording malformed events.
+    The canonical schema and generated type live in sourceos-spec. AgentTerm vendors the
+    generated TypedDict for type stability and keeps this focused fail-closed runtime
+    guard at the UI/rendering boundary.
     """
 
     errors: list[str] = []
@@ -95,25 +86,26 @@ def validate_interaction_event(event: JsonObject) -> list[str]:
     return errors
 
 
-def require_valid_interaction_event(event: JsonObject) -> None:
+def require_valid_interaction_event(event: SourceOSInteractionEvent | JsonObject) -> None:
     """Raise a ValueError if the event fails local ingest checks."""
 
     errors = validate_interaction_event(event)
     if errors:
         raise ValueError("; ".join(errors))
 
 
-def render_interaction_event(event: JsonObject) -> str:
+def render_interaction_event(event: SourceOSInteractionEvent) -> str:
     """Render a SourceOSInteractionEvent as an operator-readable governance trace."""
 
     require_valid_interaction_event(event)
 
-    surface = _object(event, "surface")
-    session = _object(event, "session")
-    actor = _object(event, "actor")
-    task = _nullable_object(event, "task")
-    steering = _nullable_object(event, "steeringIntent")
-    governance = _object(event, "governanceTrace")
+    event_obj = cast(JsonObject, event)
+    surface = _object(event_obj, "surface")
+    session = _object(event_obj, "session")
+    actor = _object(event_obj, "actor")
+    task = _nullable_object(event_obj, "task")
+    steering = _nullable_object(event_obj, "steeringIntent")
+    governance = _object(event_obj, "governanceTrace")
 
     lines = [
         "SourceOS interaction event",
@@ -178,9 +170,9 @@ def render_interaction_event(event: JsonObject) -> str:
     _append_list(lines, "grants", governance.get("grantRefs"))
     _append_list(lines, "context_packs", governance.get("contextPackRefs"))
     _append_list(lines, "evidence", governance.get("evidenceRefs"))
-    _append_list(lines, "redactions", event.get("redactionRefs"))
+    _append_list(lines, "redactions", event_obj.get("redactionRefs"))
 
-    payload = _nullable_object(event, "payload")
+    payload = _nullable_object(event_obj, "payload")
     if payload and isinstance(payload.get("summary"), str):
         lines.extend(["  payload:", f"    summary: {payload['summary']}"])
     elif event.get("payloadMode"):
@@ -190,7 +182,7 @@ def render_interaction_event(event: JsonObject) -> str:
 
 
 def interaction_to_agent_term_event(
-    event: JsonObject,
+    event: SourceOSInteractionEvent,
     *,
     channel: str = "!sourceos-interaction",
     sender: str = "@agent-term",
@@ -199,13 +191,14 @@ def interaction_to_agent_term_event(
 
     require_valid_interaction_event(event)
 
-    surface = _object(event, "surface")
-    session = _object(event, "session")
-    governance = _object(event, "governanceTrace")
-    task = _nullable_object(event, "task")
-    payload = _nullable_object(event, "payload")
+    event_obj = cast(JsonObject, event)
+    surface = _object(event_obj, "surface")
+    session = _object(event_obj, "session")
+    governance = _object(event_obj, "governanceTrace")
+    task = _nullable_object(event_obj, "task")
+    payload = _nullable_object(event_obj, "payload")
 
-    body = _summary_body(event, payload, task)
+    body = _summary_body(event_obj, payload, task)
     thread_id = session.get("threadRef") if isinstance(session.get("threadRef"), str) else None
     metadata: JsonObject = {
         "sourceos_interaction_event_id": event["interactionEventId"],
@@ -216,8 +209,8 @@ def interaction_to_agent_term_event(
         "task": task,
         "governanceTrace": governance,
         "payloadMode": event["payloadMode"],
-        "sourceEventRefs": event.get("sourceEventRefs", []),
-        "redactionRefs": event.get("redactionRefs", []),
+        "sourceEventRefs": event_obj.get("sourceEventRefs", []),
+        "redactionRefs": event_obj.get("redactionRefs", []),
     }
 
     return AgentTermEvent(

tests/test_interaction.py

@@ -4,6 +4,10 @@
 
 from pathlib import Path
 
+from agent_term.contracts.sourceos.generated.sourceos_interaction_event import (
+    SOURCEOS_INTERACTION_EVENT_REQUIRED,
+    SourceOSInteractionEvent,
+)
 from agent_term.interaction import (
     interaction_to_agent_term_event,
     load_interaction_event,
@@ -15,12 +19,24 @@
 FIXTURE = Path(__file__).parent / "fixtures" / "sourceos_interaction_event.json"
 
 
+def test_generated_contract_artifact_is_importable() -> None:
+    assert "interactionEventId" in SOURCEOS_INTERACTION_EVENT_REQUIRED
+    assert "governanceTrace" in SOURCEOS_INTERACTION_EVENT_REQUIRED
+
+
 def test_fixture_passes_local_required_field_checks() -> None:
     event = load_interaction_event(FIXTURE)
 
     assert validate_interaction_event(event) == []
 
 
+def test_loaded_fixture_is_sourceos_interaction_event_typed_dict() -> None:
+    event: SourceOSInteractionEvent = load_interaction_event(FIXTURE)
+
+    assert event["type"] == "SourceOSInteractionEvent"
+    assert event["governanceTrace"]["policyAdmitted"] is True
+
+
 def test_render_interaction_event_exposes_governance_trace() -> None:
     event = load_interaction_event(FIXTURE)