Define governed platform event envelope
Some checks failed
Dependency Audit / dependency-audit (push) Has been cancelled
Module Matrix / module-matrix (push) Has been cancelled

This commit is contained in:
2026-07-10 17:48:33 +02:00
parent 0a130962d3
commit 81f0f649b5
3 changed files with 142 additions and 3 deletions

View File

@@ -58,6 +58,16 @@ Every `PlatformEvent` has:
- `correlation_id`: stable ID for the whole request, workflow, or job. - `correlation_id`: stable ID for the whole request, workflow, or job.
- `causation_id`: the event ID or external operation ID that caused this - `causation_id`: the event ID or external operation ID that caused this
event. event.
- `actor`: optional typed actor reference, for example user, API key, system
actor, delegated actor, or installer daemon.
- `tenant`: optional tenant reference when the event is tenant-scoped.
- `subject`: optional typed subject reference for the person, organization,
account, case, campaign, or other entity the event is about.
- `resource`: optional typed resource reference for the object changed or
observed by the event.
- `classification`: payload sensitivity, currently `public`, `internal`,
`confidential`, or `restricted`.
- `payload`: JSON-serializable module-owned event details.
The FastAPI app factory creates an event context for every request. It accepts The FastAPI app factory creates an event context for every request. It accepts
`X-Correlation-ID` or `X-Request-ID` when the value is a compact safe trace ID, `X-Correlation-ID` or `X-Request-ID` when the value is a compact safe trace ID,
@@ -174,7 +184,10 @@ Campaign:
- Payloads must be JSON-serializable. - Payloads must be JSON-serializable.
- Use stable IDs, not ORM objects. - Use stable IDs, not ORM objects.
- Include tenant ID when tenant-scoped. - Include tenant ID when tenant-scoped.
- Include actor/principal references only as DTOs or primitive IDs. - Include actor/principal, subject, and resource references only as typed DTOs
or primitive IDs.
- Set `classification` to the highest sensitivity needed by the envelope or
payload, not the lowest sensitivity of any individual field.
- Do not include secrets, raw message bodies, full recipient lists, or file - Do not include secrets, raw message bodies, full recipient lists, or file
content. content.
- Put large evidence in owning module storage and reference it by ID. - Put large evidence in owning module storage and reference it by ID.

View File

@@ -7,7 +7,7 @@ from contextvars import ContextVar
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timezone from datetime import datetime, timezone
import re import re
from typing import Any from typing import Any, Literal
import uuid import uuid
@@ -31,6 +31,39 @@ class EventTrace:
causation_id: str | None = None causation_id: str | None = None
EventClassification = Literal["public", "internal", "confidential", "restricted"]
@dataclass(frozen=True, slots=True)
class EventActorRef:
type: str
id: str | None = None
label: str | None = None
def to_dict(self) -> dict[str, Any]:
return _compact_dict({"type": self.type, "id": self.id, "label": self.label})
@dataclass(frozen=True, slots=True)
class EventTenantRef:
id: str
slug: str | None = None
label: str | None = None
def to_dict(self) -> dict[str, Any]:
return _compact_dict({"id": self.id, "slug": self.slug, "label": self.label})
@dataclass(frozen=True, slots=True)
class EventObjectRef:
type: str
id: str | None = None
label: str | None = None
def to_dict(self) -> dict[str, Any]:
return _compact_dict({"type": self.type, "id": self.id, "label": self.label})
_current_trace: ContextVar[EventTrace | None] = ContextVar("govoplan_event_trace", default=None) _current_trace: ContextVar[EventTrace | None] = ContextVar("govoplan_event_trace", default=None)
@@ -43,6 +76,27 @@ class PlatformEvent:
event_id: str = field(default_factory=new_event_id) event_id: str = field(default_factory=new_event_id)
correlation_id: str | None = None correlation_id: str | None = None
causation_id: str | None = None causation_id: str | None = None
actor: EventActorRef | None = None
tenant: EventTenantRef | None = None
subject: EventObjectRef | None = None
resource: EventObjectRef | None = None
classification: EventClassification = "internal"
def to_dict(self) -> dict[str, Any]:
return {
"type": self.type,
"module_id": self.module_id,
"payload": dict(self.payload),
"occurred_at": self.occurred_at.isoformat(),
"event_id": self.event_id,
"correlation_id": self.correlation_id,
"causation_id": self.causation_id,
"actor": self.actor.to_dict() if self.actor else None,
"tenant": self.tenant.to_dict() if self.tenant else None,
"subject": self.subject.to_dict() if self.subject else None,
"resource": self.resource.to_dict() if self.resource else None,
"classification": self.classification,
}
EventHandler = Callable[[PlatformEvent], None] EventHandler = Callable[[PlatformEvent], None]
@@ -90,6 +144,11 @@ def ensure_event_trace(event: PlatformEvent) -> PlatformEvent:
event_id=event.event_id, event_id=event.event_id,
correlation_id=trace.correlation_id, correlation_id=trace.correlation_id,
causation_id=trace.causation_id, causation_id=trace.causation_id,
actor=event.actor,
tenant=event.tenant,
subject=event.subject,
resource=event.resource,
classification=event.classification,
) )
@@ -107,3 +166,7 @@ class EventBus:
handler(traced) handler(traced)
for handler in self._subscribers.get("*", ()): for handler in self._subscribers.get("*", ()):
handler(traced) handler(traced)
def _compact_dict(value: Mapping[str, Any]) -> dict[str, Any]:
return {key: item for key, item in value.items() if item is not None}

View File

@@ -9,7 +9,16 @@ from fastapi import APIRouter
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from govoplan_core.audit.logging import audit_event, audit_operation_context from govoplan_core.audit.logging import audit_event, audit_operation_context
from govoplan_core.core.events import EventBus, PlatformEvent, current_event_trace, event_context, normalize_trace_id from govoplan_core.core.events import (
EventActorRef,
EventBus,
EventObjectRef,
EventTenantRef,
PlatformEvent,
current_event_trace,
event_context,
normalize_trace_id,
)
from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.db.base import Base from govoplan_core.db.base import Base
from govoplan_core.server.fastapi import create_govoplan_app from govoplan_core.server.fastapi import create_govoplan_app
@@ -37,6 +46,60 @@ class CoreEventTests(unittest.TestCase):
self.assertEqual("corr-1", child.correlation_id) self.assertEqual("corr-1", child.correlation_id)
self.assertEqual(parent.event_id, child.causation_id) self.assertEqual(parent.event_id, child.causation_id)
def test_platform_event_serializes_governed_envelope_fields(self) -> None:
event = PlatformEvent(
type="case.permit.updated",
module_id="cases",
payload={"status": "approved"},
event_id="event-1",
correlation_id="corr-1",
causation_id="cause-1",
actor=EventActorRef(type="user", id="user-1", label="Case worker"),
tenant=EventTenantRef(id="tenant-1", slug="city", label="City Office"),
subject=EventObjectRef(type="person", id="person-1", label="Applicant"),
resource=EventObjectRef(type="permit", id="permit-1", label="Permit"),
classification="confidential",
)
payload = event.to_dict()
self.assertEqual("case.permit.updated", payload["type"])
self.assertEqual("cases", payload["module_id"])
self.assertEqual({"status": "approved"}, payload["payload"])
self.assertEqual("event-1", payload["event_id"])
self.assertEqual("corr-1", payload["correlation_id"])
self.assertEqual("cause-1", payload["causation_id"])
self.assertEqual({"type": "user", "id": "user-1", "label": "Case worker"}, payload["actor"])
self.assertEqual({"id": "tenant-1", "slug": "city", "label": "City Office"}, payload["tenant"])
self.assertEqual({"type": "person", "id": "person-1", "label": "Applicant"}, payload["subject"])
self.assertEqual({"type": "permit", "id": "permit-1", "label": "Permit"}, payload["resource"])
self.assertEqual("confidential", payload["classification"])
def test_event_bus_preserves_envelope_fields_when_adding_trace(self) -> None:
bus = EventBus()
seen: list[PlatformEvent] = []
event = PlatformEvent(
type="files.file.uploaded",
module_id="files",
actor=EventActorRef(type="api_key", id="key-1"),
tenant=EventTenantRef(id="tenant-1"),
subject=EventObjectRef(type="file", id="file-1"),
resource=EventObjectRef(type="folder", id="folder-1"),
classification="restricted",
)
bus.subscribe("files.file.uploaded", seen.append)
bus.publish(event)
self.assertEqual(1, len(seen))
traced = seen[0]
self.assertIsNotNone(traced.correlation_id)
self.assertEqual(event.actor, traced.actor)
self.assertEqual(event.tenant, traced.tenant)
self.assertEqual(event.subject, traced.subject)
self.assertEqual(event.resource, traced.resource)
self.assertEqual("restricted", traced.classification)
def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None: def test_request_correlation_middleware_sets_event_context_and_response_header(self) -> None:
router = APIRouter() router = APIRouter()