Add nested segment timing logs

README.md

@@ -48,6 +48,19 @@ Multiple destinations can be enabled with a comma-separated list, for example
 Logging uses Application Default Credentials and requires permission to create
 log entries, typically through `roles/logging.logWriter`.
 
+Request and operation logs include two timing views:
+
+- `timings_ms` and `timing_counts` are flat inclusive aggregates by segment
+  name, intended for quick scanning and compatibility with existing log
+  queries.
+- `segment_tree` is an ordered nested view of segment occurrences. Repeated
+  sibling segments are preserved as separate entries, and safe scalar segment
+  attributes are included so callers can distinguish settings such as
+  `simulation_kind=baseline` versus `simulation_kind=reform`.
+
+Core structured log fields take precedence over caller-provided attributes with
+the same keys.
+
 On runtimes that do not provide Application Default Credentials, set
 `GCP_CREDENTIALS_JSON` to a service account JSON document. The Google Cloud
 Logging destination will materialize it into a temporary credentials file and

changelog.d/15.added.md

@@ -0,0 +1,2 @@
+Add ordered nested segment trees to request and operation structured logs.
+Core structured log fields now take precedence over caller-provided attributes.

policyengine_observability/context.py

@@ -24,6 +24,28 @@ def as_dict(self) -> dict[str, Any]:
         }
 
 
+@dataclass
+class SegmentTimingNode:
+    sequence: int
+    name: str
+    attrs: dict[str, Any] = field(default_factory=dict)
+    duration_ms: float | None = None
+    children: list[SegmentTimingNode] = field(default_factory=list)
+
+    def as_dict(self) -> dict[str, Any]:
+        record: dict[str, Any] = {
+            "sequence": self.sequence,
+            "name": self.name,
+        }
+        if self.attrs:
+            record["attrs"] = dict(self.attrs)
+        if self.duration_ms is not None:
+            record["duration_ms"] = round(self.duration_ms, 3)
+        if self.children:
+            record["children"] = [child.as_dict() for child in self.children]
+        return record
+
+
 @dataclass
 class OperationObservabilityContext:
     config: ObservabilityConfig
@@ -32,6 +54,8 @@ class OperationObservabilityContext:
     attributes: dict[str, Any] = field(default_factory=dict)
     timings_ms: dict[str, float] = field(default_factory=dict)
     timing_counts: dict[str, int] = field(default_factory=dict)
+    segment_tree: list[SegmentTimingNode] = field(default_factory=list)
+    segment_sequence: list[int] = field(default_factory=lambda: [0])
     emit_log: bool = True
     record_metric: bool = True
     started_at: float = field(default_factory=time.perf_counter)
@@ -94,6 +118,7 @@ def as_log_record(
     ) -> dict[str, Any]:
         event = "operation_failed" if self.error else "operation_completed"
         return {
+            **self.attributes,
             "schema_version": "policyengine.observability.operation.v1",
             "event": event,
             "service_name": self.config.service_name,
@@ -107,7 +132,7 @@ def as_log_record(
             "duration_ms": round(self.duration_seconds() * 1000, 3),
             "timings_ms": dict(self.timings_ms),
             "timing_counts": dict(self.timing_counts),
-            **self.attributes,
+            "segment_tree": [node.as_dict() for node in self.segment_tree],
             "error": self.error.as_dict() if self.error else None,
         }
 
@@ -129,6 +154,8 @@ class RequestObservabilityContext:
     attributes: dict[str, Any] = field(default_factory=dict)
     timings_ms: dict[str, float] = field(default_factory=dict)
     timing_counts: dict[str, int] = field(default_factory=dict)
+    segment_tree: list[SegmentTimingNode] = field(default_factory=list)
+    segment_sequence: list[int] = field(default_factory=lambda: [0])
     status_code: int | None = None
     error: ErrorRecord | None = None
     emitted: bool = False
@@ -206,6 +233,8 @@ def as_log_record(
         )
         status_code = self.status_code or (500 if self.error else None)
         return {
+            **self.inbound,
+            **self.attributes,
             "schema_version": "policyengine.observability.request.v1",
             "event": event,
             "service_name": self.config.service_name,
@@ -222,10 +251,9 @@ def as_log_record(
             "endpoint": self.endpoint,
             "status_code": status_code,
             "duration_ms": round(self.duration_seconds() * 1000, 3),
-            **self.inbound,
             "timings_ms": dict(self.timings_ms),
             "timing_counts": dict(self.timing_counts),
-            **self.attributes,
+            "segment_tree": [node.as_dict() for node in self.segment_tree],
             "error": self.error.as_dict() if self.error else None,
         }