“

Series: S5 Intelligent Decisions · Article 11 | Level: Advanced | Reading Time: 20 min

Decision Trace Chain: End-to-End Traceability from Input to Execution

#TL;DR

An enterprise decision system must answer one core question: "How was this decision made?" coomia-dip builds an end-to-end Decision Trace Chain from raw data input to final execution results, based on the OpenTelemetry distributed tracing standard, linking all intermediate states across the Sense, Think, Decide, and Act stages into a complete causal chain. This article dissects the trace chain data model, Span design, storage architecture, and query API.

#1. Why End-to-End Tracing Is Essential

#1.1 Regulatory Requirements for Decision Auditing

Code

Regulatory Compliance Requirements:

  GDPR (EU)           Financial Regulation    Medical Compliance
  +--------------+    +--------------+        +--------------+
  | Data subjects|    | Model explain-|       | Clinical      |
  | have right to|    | ability req.  |       | decisions must|
  | understand   |    |              |        | be recorded   |
  | automated    |    |              |        |               |
  | decisions    |    |              |        |               |
  +--------------+    +--------------+        +--------------+
         |                  |                       |
         +--------+---------+-----------------------+
                  |
                  v
         +------------------+
         | Complete Decision |
         | Trace Chain       |
         +------------------+

#1.2 Trace Chain Coverage

Code

Decision Trace Chain Full Pipeline:

  Data Input -> Feature Calc -> Rule Match -> ML Infer -> Fusion
     |             |              |            |           |
     v             v              v            v           v
  [Span]        [Span]         [Span]       [Span]      [Span]
     |             |              |            |           |
     +-------------+--------------+------------+-----------+
                              |
                              v
                      -> Approval -> Action Exec -> Result Feedback
                          |            |              |
                          v            v              v
                        [Span]       [Span]         [Span]
                          |            |              |
                          +------------+--------------+
                                      |
                                      v
                              TraceID: unified linkage

#2. Trace Data Model

#2.1 Core Models

Python

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
import uuid


class TracePhase(Enum):
    """Trace phase"""
    SENSE = "sense"
    THINK = "think"
    DECIDE = "decide"
    ACT = "act"
    APPROVE = "approve"


class SpanKind(Enum):
    """Span type"""
    DATA_INPUT = "data_input"
    FEATURE_COMPUTE = "feature_compute"
    RULE_EVALUATION = "rule_evaluation"
    ML_INFERENCE = "ml_inference"
    FUSION = "fusion"
    DECISION = "decision"
    APPROVAL = "approval"
    ACTION_EXECUTION = "action_execution"
    NOTIFICATION = "notification"


@dataclass
class DecisionTrace:
    """Decision trace"""
    trace_id: str
    decision_id: str
    domain: str
    initiator: str
    start_time: datetime
    end_time: datetime | None = None
    status: str = "in_progress"
    spans: list[TraceSpan] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

    @staticmethod
    def new(domain: str, initiator: str) -> DecisionTrace:
        trace_id = str(uuid.uuid4()).replace("-", "")[:32]
        return DecisionTrace(
            trace_id=trace_id,
            decision_id=f"dec-{trace_id[:12]}",
            domain=domain,
            initiator=initiator,
            start_time=datetime.utcnow(),
        )


@dataclass
class TraceSpan:
    """Trace span"""
    span_id: str
    parent_span_id: str | None
    trace_id: str
    phase: TracePhase
    kind: SpanKind
    name: str
    start_time: datetime
    end_time: datetime | None = None
    status: str = "ok"
    attributes: dict[str, Any] = field(default_factory=dict)
    events: list[SpanEvent] = field(default_factory=list)
    links: list[SpanLink] = field(default_factory=list)

    @staticmethod
    def new(trace_id: str, phase: TracePhase, kind: SpanKind,
            name: str, parent_id: str | None = None) -> TraceSpan:
        return TraceSpan(
            span_id=str(uuid.uuid4()).replace("-", "")[:16],
            parent_span_id=parent_id,
            trace_id=trace_id,
            phase=phase,
            kind=kind,
            name=name,
            start_time=datetime.utcnow(),
        )


@dataclass
class SpanEvent:
    """Span event"""
    name: str
    timestamp: datetime
    attributes: dict[str, Any] = field(default_factory=dict)


@dataclass
class SpanLink:
    """Span link"""
    linked_trace_id: str
    linked_span_id: str
    relationship: str = "caused_by"

#2.2 Trace Context Propagation

Python

from contextvars import ContextVar

_current_trace: ContextVar[DecisionTrace | None] = ContextVar(
    "current_trace", default=None
)
_current_span: ContextVar[TraceSpan | None] = ContextVar(
    "current_span", default=None
)


class TraceContext:
    """Trace context management"""

    @staticmethod
    def start_trace(domain: str, initiator: str) -> DecisionTrace:
        trace = DecisionTrace.new(domain, initiator)
        _current_trace.set(trace)
        return trace

    @staticmethod
    def current_trace() -> DecisionTrace | None:
        return _current_trace.get()

    @staticmethod
    def start_span(phase: TracePhase, kind: SpanKind,
                   name: str) -> TraceSpan:
        trace = _current_trace.get()
        if trace is None:
            raise RuntimeError("No active trace")

        parent = _current_span.get()
        span = TraceSpan.new(
            trace_id=trace.trace_id,
            phase=phase,
            kind=kind,
            name=name,
            parent_id=parent.span_id if parent else None,
        )
        trace.spans.append(span)
        _current_span.set(span)
        return span

    @staticmethod
    def end_span(span: TraceSpan, status: str = "ok") -> None:
        span.end_time = datetime.utcnow()
        span.status = status
        trace = _current_trace.get()
        if trace and span.parent_span_id:
            parent = next(
                (s for s in trace.spans if s.span_id == span.parent_span_id),
                None
            )
            _current_span.set(parent)
        else:
            _current_span.set(None)

    @staticmethod
    def add_event(name: str, **attributes) -> None:
        span = _current_span.get()
        if span:
            span.events.append(SpanEvent(
                name=name,
                timestamp=datetime.utcnow(),
                attributes=attributes,
            ))

    @staticmethod
    def set_attribute(key: str, value: Any) -> None:
        span = _current_span.get()
        if span:
            span.attributes[key] = value

#3. Phase-Specific Span Design

#3.1 Sense Phase

Python

class SenseTracer:
    """Sense phase tracing"""

    @staticmethod
    def trace_data_input(source: str, record_count: int,
                          schema: dict) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.SENSE, SpanKind.DATA_INPUT,
            f"data_input.{source}"
        )
        TraceContext.set_attribute("source.type", source)
        TraceContext.set_attribute("source.record_count", record_count)
        TraceContext.set_attribute("source.schema", str(schema))
        return span

    @staticmethod
    def trace_feature_compute(features: dict[str, Any]) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.SENSE, SpanKind.FEATURE_COMPUTE,
            "feature_computation"
        )
        TraceContext.set_attribute("features.count", len(features))
        for name, value in features.items():
            TraceContext.set_attribute(f"feature.{name}", str(value))
        return span

#3.2 Think Phase

Python

class ThinkTracer:
    """Think phase tracing"""

    @staticmethod
    def trace_rule_evaluation(rule_set: str,
                               rules_count: int) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.THINK, SpanKind.RULE_EVALUATION,
            f"rule_eval.{rule_set}"
        )
        TraceContext.set_attribute("rules.set", rule_set)
        TraceContext.set_attribute("rules.total_count", rules_count)
        return span

    @staticmethod
    def trace_rule_match(rule_id: str, matched: bool,
                          conditions: list[dict]) -> None:
        TraceContext.add_event(
            "rule_match",
            rule_id=rule_id,
            matched=matched,
            conditions=str(conditions),
        )

    @staticmethod
    def trace_ml_inference(model_name: str, model_version: str,
                            prediction: str,
                            confidence: float) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.THINK, SpanKind.ML_INFERENCE,
            f"ml_inference.{model_name}"
        )
        TraceContext.set_attribute("model.name", model_name)
        TraceContext.set_attribute("model.version", model_version)
        TraceContext.set_attribute("model.prediction", prediction)
        TraceContext.set_attribute("model.confidence", confidence)
        return span

#3.3 Decide Phase

Python

class DecideTracer:
    """Decide phase tracing"""

    @staticmethod
    def trace_fusion(method: str, rule_result: str,
                      ml_result: str, final: str,
                      confidence: float) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.DECIDE, SpanKind.FUSION,
            f"fusion.{method}"
        )
        TraceContext.set_attribute("fusion.method", method)
        TraceContext.set_attribute("fusion.rule_result", rule_result)
        TraceContext.set_attribute("fusion.ml_result", ml_result)
        TraceContext.set_attribute("fusion.final_decision", final)
        TraceContext.set_attribute("fusion.confidence", confidence)
        return span

    @staticmethod
    def trace_decision(decision: str, confidence: float,
                        needs_approval: bool) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.DECIDE, SpanKind.DECISION,
            "decision_output"
        )
        TraceContext.set_attribute("decision.result", decision)
        TraceContext.set_attribute("decision.confidence", confidence)
        TraceContext.set_attribute("decision.needs_approval", needs_approval)
        return span

#3.4 Act Phase

Python

class ActTracer:
    """Act phase tracing"""

    @staticmethod
    def trace_action_execution(action_type: str,
                                target: str) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.ACT, SpanKind.ACTION_EXECUTION,
            f"action.{action_type}"
        )
        TraceContext.set_attribute("action.type", action_type)
        TraceContext.set_attribute("action.target", target)
        return span

    @staticmethod
    def trace_notification(channel: str, recipient: str,
                            template: str) -> TraceSpan:
        span = TraceContext.start_span(
            TracePhase.ACT, SpanKind.NOTIFICATION,
            f"notify.{channel}"
        )
        TraceContext.set_attribute("notification.channel", channel)
        TraceContext.set_attribute("notification.recipient", recipient)
        TraceContext.set_attribute("notification.template", template)
        return span

#4. Trace Storage

#4.1 Storage Architecture

Code

Trace Data Storage Tiers:

  Real-time (< 24h)      Near-line (1-30d)     Offline (> 30d)
  +-------------+        +-------------+       +-------------+
  | PostgreSQL  |        | Iceberg     |       | Iceberg     |
  | (JSONB)     |        | (Parquet)   |       | (Archive)   |
  +------+------+        +------+------+       +------+------+
         |                      |                     |
         v                      v                     v
  Index: trace_id         Partition: day/domain  Partition: month
  Index: decision_id      Compression: Snappy    Retention: 7yr
  Index: timestamp        TTL: 30 days           Cold storage

#4.2 Storage Implementation

Python

import json
from datetime import datetime


class TraceStore:
    """Trace data store"""

    def __init__(self, db_pool, iceberg_catalog):
        self._db = db_pool
        self._iceberg = iceberg_catalog

    async def save_trace(self, trace: DecisionTrace) -> None:
        async with self._db.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO decision_traces
                (trace_id, decision_id, domain, initiator,
                 start_time, end_time, status, spans, metadata)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                ON CONFLICT (trace_id) DO UPDATE SET
                    end_time = $6, status = $7, spans = $8
                """,
                trace.trace_id,
                trace.decision_id,
                trace.domain,
                trace.initiator,
                trace.start_time,
                trace.end_time,
                trace.status,
                json.dumps([self._span_to_dict(s) for s in trace.spans]),
                json.dumps(trace.metadata),
            )

    async def get_trace(self, trace_id: str) -> DecisionTrace | None:
        async with self._db.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM decision_traces WHERE trace_id = $1",
                trace_id,
            )
            if row:
                return self._row_to_trace(row)
            return None

    async def query_traces(self, domain: str | None = None,
                            start: datetime | None = None,
                            end: datetime | None = None,
                            status: str | None = None,
                            limit: int = 100) -> list[DecisionTrace]:
        conditions = []
        params = []
        idx = 1

        if domain:
            conditions.append(f"domain = ${idx}")
            params.append(domain)
            idx += 1

        if start:
            conditions.append(f"start_time >= ${idx}")
            params.append(start)
            idx += 1

        if end:
            conditions.append(f"start_time <= ${idx}")
            params.append(end)
            idx += 1

        if status:
            conditions.append(f"status = ${idx}")
            params.append(status)
            idx += 1

        where = " AND ".join(conditions) if conditions else "TRUE"
        query = f"""
            SELECT * FROM decision_traces
            WHERE {where}
            ORDER BY start_time DESC
            LIMIT {limit}
        """

        async with self._db.acquire() as conn:
            rows = await conn.fetch(query, *params)
            return [self._row_to_trace(r) for r in rows]

    def _span_to_dict(self, span: TraceSpan) -> dict:
        return {
            "span_id": span.span_id,
            "parent_span_id": span.parent_span_id,
            "phase": span.phase.value,
            "kind": span.kind.value,
            "name": span.name,
            "start_time": span.start_time.isoformat(),
            "end_time": span.end_time.isoformat() if span.end_time else None,
            "status": span.status,
            "attributes": span.attributes,
            "events": [
                {
                    "name": e.name,
                    "timestamp": e.timestamp.isoformat(),
                    "attributes": e.attributes,
                }
                for e in span.events
            ],
        }

    def _row_to_trace(self, row) -> DecisionTrace:
        spans_data = json.loads(row["spans"])
        return DecisionTrace(
            trace_id=row["trace_id"],
            decision_id=row["decision_id"],
            domain=row["domain"],
            initiator=row["initiator"],
            start_time=row["start_time"],
            end_time=row["end_time"],
            status=row["status"],
            spans=[self._dict_to_span(s, row["trace_id"]) for s in spans_data],
            metadata=json.loads(row["metadata"]),
        )

    def _dict_to_span(self, data: dict, trace_id: str) -> TraceSpan:
        return TraceSpan(
            span_id=data["span_id"],
            parent_span_id=data.get("parent_span_id"),
            trace_id=trace_id,
            phase=TracePhase(data["phase"]),
            kind=SpanKind(data["kind"]),
            name=data["name"],
            start_time=datetime.fromisoformat(data["start_time"]),
            end_time=(
                datetime.fromisoformat(data["end_time"])
                if data.get("end_time") else None
            ),
            status=data.get("status", "ok"),
            attributes=data.get("attributes", {}),
        )

#5. gRPC Query Service

#5.1 Protobuf Definition

PROTOBUF

syntax = "proto3";
package onto.trace.v1;

service TraceService {
    rpc GetTrace(GetTraceRequest) returns (TraceResponse);
    rpc QueryTraces(QueryRequest) returns (QueryResponse);
    rpc GetTraceTimeline(TimelineRequest) returns (TimelineResponse);
    rpc GetTraceGraph(GraphRequest) returns (GraphResponse);
}

message GetTraceRequest {
    string trace_id = 1;
}

message TraceResponse {
    string trace_id = 1;
    string decision_id = 2;
    string domain = 3;
    string status = 4;
    repeated SpanData spans = 5;
    string timeline_ascii = 6;
}

message SpanData {
    string span_id = 1;
    string parent_span_id = 2;
    string phase = 3;
    string kind = 4;
    string name = 5;
    int64 start_time_ms = 6;
    int64 duration_ms = 7;
    string status = 8;
    map<string, string> attributes = 9;
}

#6. Trace Visualization

#6.1 Timeline View

Code

Decision Trace Timeline (trace_id: a1b2c3d4e5f6)
=============================================

  Time(ms)   0     10    20    30    40    50    60    70

  SENSE
  | data_input.api     [====]                              12ms
  | feature_compute    [  ====]                            15ms
  |
  THINK
  | rule_eval.credit   [      ====]                         8ms
  | ml_inference.v3    [      ==========]                   22ms
  |
  DECIDE
  | fusion.weighted    [                  ===]               6ms
  | decision_output    [                     ==]             4ms
  |
  APPROVE
  | approval.submit    [                       ==]           3ms
  | approval.wait      [                         ........]  pending
  |
  ACT
  + (awaiting approval)

  Total: 70ms (excluding approval wait)
  Decision: conditional_approve (confidence: 0.72)

#6.2 Causal Graph View

Code

Decision Causal Graph:

  [data_input.api] --> [feature_compute]
                              |
                    +---------+---------+
                    v         v         v
            [rule_eval]  [ml_inference]
                    |         |
                    +----+----+
                         v
                    [fusion.weighted]
                         |
                         v
                  [decision_output]
                         |
                    +----+----+
                    v         v
             [approval]  [notification]
                    |
                    v
              [action_exec]

#7. OpenTelemetry Integration

#7.1 Exporter

Python

from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)


class OTelTraceExporter:
    """Export decision traces to OpenTelemetry"""

    def __init__(self, endpoint: str = "localhost:4317"):
        provider = TracerProvider()
        exporter = OTLPSpanExporter(endpoint=endpoint)
        provider.add_span_processor(BatchSpanProcessor(exporter))
        otel_trace.set_tracer_provider(provider)
        self._tracer = otel_trace.get_tracer("coomia-dip-decision")

    def export_trace(self, trace: DecisionTrace) -> None:
        span_map: dict[str, Any] = {}

        root_spans = [s for s in trace.spans if s.parent_span_id is None]
        child_spans = [s for s in trace.spans if s.parent_span_id is not None]

        for span in root_spans:
            otel_span = self._create_otel_span(span, None)
            span_map[span.span_id] = otel_span

        for span in child_spans:
            parent_ctx = span_map.get(span.parent_span_id)
            otel_span = self._create_otel_span(span, parent_ctx)
            span_map[span.span_id] = otel_span

    def _create_otel_span(self, span: TraceSpan,
                           parent_context) -> Any:
        ctx = otel_trace.set_span_in_context(parent_context) if parent_context else None
        with self._tracer.start_as_current_span(
            span.name,
            context=ctx,
            attributes={
                "decision.phase": span.phase.value,
                "decision.kind": span.kind.value,
                **{k: str(v) for k, v in span.attributes.items()},
            },
        ) as otel_span:
            for event in span.events:
                otel_span.add_event(
                    event.name,
                    attributes={k: str(v) for k, v in event.attributes.items()},
                )
            return otel_span

#8. Trace Analysis

#8.1 Trace Statistics

Python

class TraceAnalyzer:
    """Trace analyzer"""

    def analyze(self, trace: DecisionTrace) -> dict:
        spans = trace.spans
        total_duration = (
            (trace.end_time - trace.start_time).total_seconds() * 1000
            if trace.end_time else 0
        )

        phase_durations: dict[str, float] = {}
        for span in spans:
            if span.end_time:
                duration = (span.end_time - span.start_time).total_seconds() * 1000
                phase = span.phase.value
                phase_durations[phase] = phase_durations.get(phase, 0) + duration

        bottleneck = max(phase_durations.items(), key=lambda x: x[1]) if phase_durations else ("none", 0)

        return {
            "trace_id": trace.trace_id,
            "total_duration_ms": total_duration,
            "span_count": len(spans),
            "phase_durations_ms": phase_durations,
            "bottleneck_phase": bottleneck[0],
            "bottleneck_ms": bottleneck[1],
            "error_spans": [s.name for s in spans if s.status == "error"],
            "decision": next(
                (s.attributes.get("decision.result")
                 for s in spans if s.kind == SpanKind.DECISION),
                "unknown"
            ),
        }

    def analyze_batch(self, traces: list[DecisionTrace]) -> dict:
        analyses = [self.analyze(t) for t in traces]
        durations = [a["total_duration_ms"] for a in analyses if a["total_duration_ms"] > 0]

        if not durations:
            return {"count": 0}

        durations.sort()
        return {
            "count": len(analyses),
            "avg_duration_ms": sum(durations) / len(durations),
            "p50_duration_ms": durations[len(durations) // 2],
            "p99_duration_ms": durations[int(len(durations) * 0.99)],
            "error_rate": sum(1 for a in analyses if a["error_spans"]) / len(analyses),
            "decision_distribution": self._count_decisions(analyses),
        }

    def _count_decisions(self, analyses: list[dict]) -> dict[str, int]:
        dist: dict[str, int] = {}
        for a in analyses:
            d = a.get("decision", "unknown")
            dist[d] = dist.get(d, 0) + 1
        return dist

#9. Security and Privacy

#9.1 Data Sanitization

Python

class TraceSanitizer:
    """Trace data sanitizer"""

    SENSITIVE_KEYS = {
        "ssn", "id_number", "phone", "email",
        "credit_card", "password", "secret",
    }

    @classmethod
    def sanitize(cls, trace: DecisionTrace) -> DecisionTrace:
        for span in trace.spans:
            sanitized_attrs = {}
            for key, value in span.attributes.items():
                if any(sk in key.lower() for sk in cls.SENSITIVE_KEYS):
                    sanitized_attrs[key] = cls._mask(str(value))
                else:
                    sanitized_attrs[key] = value
            span.attributes = sanitized_attrs

            for event in span.events:
                for key in list(event.attributes.keys()):
                    if any(sk in key.lower() for sk in cls.SENSITIVE_KEYS):
                        event.attributes[key] = cls._mask(
                            str(event.attributes[key])
                        )

        return trace

    @staticmethod
    def _mask(value: str) -> str:
        if len(value) <= 4:
            return "****"
        return value[:2] + "*" * (len(value) - 4) + value[-2:]

#10. Practical Example

Python

# Complete decision tracing flow

# 1. Start trace
trace = TraceContext.start_trace(domain="credit", initiator="api-gateway")

# 2. Sense phase
span = SenseTracer.trace_data_input("api", record_count=1, schema={"credit_score": "int"})
TraceContext.end_span(span)

span = SenseTracer.trace_feature_compute({"credit_score": 620, "debt_ratio": 0.45})
TraceContext.end_span(span)

# 3. Think phase
span = ThinkTracer.trace_rule_evaluation("credit_rules", rules_count=5)
ThinkTracer.trace_rule_match("CR-001", matched=False, conditions=[{"credit_score >= 700": False}])
ThinkTracer.trace_rule_match("CR-003", matched=True, conditions=[{"credit_score >= 550": True}])
TraceContext.end_span(span)

span = ThinkTracer.trace_ml_inference("credit_v3", "3.2.1", "conditional_approve", 0.68)
TraceContext.end_span(span)

# 4. Decide phase
span = DecideTracer.trace_fusion("weighted", "conditional_approve", "conditional_approve",
                                  "conditional_approve", 0.72)
TraceContext.end_span(span)

span = DecideTracer.trace_decision("conditional_approve", 0.72, needs_approval=True)
TraceContext.end_span(span)

# 5. Save
trace_obj = TraceContext.current_trace()
trace_obj.end_time = datetime.utcnow()
trace_obj.status = "completed"
await trace_store.save_trace(trace_obj)

# 6. Analyze
analyzer = TraceAnalyzer()
analysis = analyzer.analyze(trace_obj)
# {
#   "total_duration_ms": 52.3,
#   "span_count": 6,
#   "bottleneck_phase": "think",
#   "bottleneck_ms": 30.0,
#   "decision": "conditional_approve",
# }

#Key Takeaways

End-to-end trace chain links all intermediate states across Sense-Think-Decide-Act stages
Layered Span design organizes trace data by phase and type for fine-grained analysis
ContextVar propagation enables zero-intrusion trace context passing
Three-tier storage balances query performance with storage cost across real-time/near-line/offline layers
OpenTelemetry compatible exports to standard OTel ecosystem (Jaeger/Zipkin)
Data sanitization automatically identifies and masks sensitive fields for privacy compliance
Trace analysis supports single and batch analysis to identify performance bottlenecks and decision distributions

#Next Article

Next up: S5-12 Action Engine: Unified Orchestration of 10 Executor Types dives deep into how the coomia-dip ActionEngine unifies gRPC, HTTP, message queue, and 7 other executor types.

tags: #decision-trace #opentelemetry #distributed-tracing #audit #compliance #span #coomia-dip