Using OpenChainGraph with OpenTelemetry

/**
 * artifactToOtelSpan(artifact, options)
 *   artifact       — a completed OCG v0.3 artifact object
 *   options.rootHash — hex string: execution_hash of the chain root artifact
 *                     (if omitted, this artifact is treated as root)
 *   options.durationMs — number: override the default 1ms span duration
 *
 * Returns an OTLP-compatible span object for use in a ResourceSpans payload.
 */
function artifactToOtelSpan(artifact, options = {}) {
  const rootHash   = options.rootHash ?? artifact.execution_hash;
  const durationMs = options.durationMs ?? 1;
  const isRoot     = rootHash === artifact.execution_hash;

  // Derive OTel IDs from execution hashes (hex substring truncation)
  const traceId    = rootHash.slice(0, 32);   // 16 bytes = 32 hex chars
  const spanId     = artifact.execution_hash.slice(0, 16);  // 8 bytes = 16 hex chars
  const parentSpanId = (!isRoot && artifact.parent_hashes?.length)
    ? artifact.parent_hashes[0].slice(0, 16)
    : undefined;

  // Derive OTel status from compliance_flags[]
  const flags = artifact.compliance_flags ?? [];
  let statusCode = 0;  // UNSET
  if (flags.every(f => f.status === "PASS"))       statusCode = 1;  // OK
  else if (flags.some(f => f.status === "FAIL"))   statusCode = 2;  // ERROR

  // Timestamps: nanosecond precision
  const startNs = BigInt(new Date(artifact.generated_at).getTime()) * 1_000_000n;
  const endNs   = startNs + BigInt(durationMs) * 1_000_000n;

  // Build span name from mandate_type + tool_id
  const spanName = `${artifact.mandate_type ?? "unknown"}/${artifact.tool_id ?? "unknown"}`;

  // Collect all secondary parent hashes as an attribute
  const secondaryParents = (artifact.parent_hashes ?? []).slice(1);

  // Build the span object (OTLP JSON schema)
  const span = {
    traceId,
    spanId,
    name: spanName,
    kind: 3,  // SPAN_KIND_CLIENT — deterministic tool invocation
    startTimeUnixNano: startNs.toString(),
    endTimeUnixNano:   endNs.toString(),
    status: {
      code: statusCode,
      ...(statusCode === 2 && {
        message: flags.filter(f => f.status === "FAIL").map(f => f.framework).join(", ")
      })
    },
    attributes: [
      { key: "ocg.tool_id",          value: { stringValue: artifact.tool_id ?? "" } },
      { key: "ocg.tool_version",      value: { stringValue: artifact.tool_version ?? "" } },
      { key: "ocg.ap2_version",       value: { stringValue: artifact.ap2_version ?? "" } },
      { key: "ocg.mandate_type",      value: { stringValue: artifact.mandate_type ?? "" } },
      { key: "ocg.execution_hash",    value: { stringValue: artifact.execution_hash ?? "" } },
      { key: "ocg.generated_at",      value: { stringValue: artifact.generated_at ?? "" } },
      { key: "ocg.zero_pii_verified", value: { boolValue: artifact.audit_signature?.audit_signature?.zero_pii_verified ?? true } },
      ...(artifact.audit_signature?.signatures?.[0]?.keyid
        ? [{ key: "ocg.signer_keyid", value: { stringValue: artifact.audit_signature.signatures[0].keyid } }]
        : []),
      ...(artifact.audit_signature?.signatures?.[0]?.sig
        ? [{ key: "ocg.audit_sig",    value: { stringValue: artifact.audit_signature.signatures[0].sig } }]
        : []),
      ...(secondaryParents.length
        ? [{ key: "ocg.secondary_parent_hashes", value: { stringValue: secondaryParents.join(",") } }]
        : []),
      ...(flags.length
        ? [{ key: "ocg.compliance_flags", value: { stringValue: JSON.stringify(flags) } }]
        : []),
    ]
  };

  // Attach parentSpanId only on non-root spans
  if (parentSpanId) span.parentSpanId = parentSpanId;

  return span;
}

/**
 * chainToOtlpPayload(artifacts, options)
 *   artifacts — array of OCG artifacts in any order; root is auto-detected
 *   options.serviceName — string for the OTel Resource (default: "ocg-chain")
 *   options.durationMs  — optional per-span duration override
 *
 * Returns a complete OTLP ResourceSpans payload, ready for POST to /v1/traces.
 */
function chainToOtlpPayload(artifacts, options = {}) {
  const serviceName = options.serviceName ?? "ocg-chain";

  // Auto-detect root: artifact whose execution_hash doesn't appear in any parent_hashes[]
  const allParentHashes = new Set(
    artifacts.flatMap(a => a.parent_hashes ?? [])
  );
  const rootArtifact = artifacts.find(a => !allParentHashes.has(a.execution_hash));
  const rootHash = rootArtifact?.execution_hash ?? artifacts[0].execution_hash;

  const spans = artifacts.map(a => artifactToOtelSpan(a, { ...options, rootHash }));

  return {
    resourceSpans: [{
      resource: {
        attributes: [
          { key: "service.name", value: { stringValue: serviceName } },
          { key: "ocg.root_hash", value: { stringValue: rootHash } },
          { key: "ocg.chain_length", value: { intValue: artifacts.length } }
        ]
      },
      scopeSpans: [{
        scope: { name: "ocg-otel-adapter", version: "0.2.0" },
        spans
      }]
    }]
  };
}

/**
 * exportChainToOtel(artifacts, collectorUrl, options)
 *   artifacts    — array of completed OCG v0.3 artifact objects
 *   collectorUrl — base URL of the OTel Collector HTTP receiver
 *                  e.g. "http://localhost:4318" or "https://otlp.nr-data.net"
 *   options      — passed through to chainToOtlpPayload()
 *
 * Returns { ok: bool, status: number, message: string }
 */
async function exportChainToOtel(artifacts, collectorUrl, options = {}) {
  const payload = chainToOtlpPayload(artifacts, options);
  const url     = collectorUrl.replace(/\/+$/, "") + "/v1/traces";

  try {
    const res = await fetch(url, {
      method:  "POST",
      headers: {
        "Content-Type": "application/json",
        ...(options.headers ?? {})  // pass API keys / auth headers here
      },
      body: JSON.stringify(payload)
    });

    if (!res.ok) {
      const text = await res.text().catch(() => "");
      return { ok: false, status: res.status, message: text };
    }

    return { ok: true, status: res.status, message: "Spans accepted" };

  } catch (err) {
    return { ok: false, status: 0, message: err.message };
  }
}

// Usage — emit a 4-tool decision chain to a local collector:
const result = await exportChainToOtel(
  [mandateArtifact, riskArtifact, complianceArtifact, capitalArtifact],
  "http://localhost:4318",
  { serviceName: "ainumbers-payment-auth" }
);
console.log(result);  // { ok: true, status: 200, message: "Spans accepted" }

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      http:
        endpoint: "0.0.0.0:4318"
        cors:
          allowed_origins: ["*"]  # restrict to your domain in production

processors:
  batch:
    timeout: "5s"
    send_batch_size: 100
  attributes:
    actions:
      - key: "deployment.environment"
        value: "production"
        action: insert

exporters:
  datadog:
    api:
      key: "${DD_API_KEY}"
  jaeger:
    endpoint: "jaeger-collector:14250"
    tls:
      insecure: true
  otlp/tempo:
    endpoint: "tempo:4317"
    tls:
      insecure: true

service:
  pipelines:
    traces:
      receivers:  [otlp]
      processors: [batch, attributes]
      exporters:  [datadog, jaeger, otlp/tempo]

import json, requests
from datetime import datetime, timezone

def iso_to_unix_ns(iso_str: str) -> int:
    """ISO 8601 string → Unix nanoseconds (integer)."""
    dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
    return int(dt.timestamp() * 1e9)

def artifact_to_otel_span(artifact: dict, root_hash: str, duration_ms: int = 1) -> dict:
    """Convert a single OCG v0.3 artifact to an OTLP span dict."""
    execution_hash = artifact["execution_hash"]
    parent_hashes  = artifact.get("parent_hashes", [])
    is_root = (execution_hash == root_hash)

    trace_id   = root_hash[:32]
    span_id    = execution_hash[:16]
    parent_id  = parent_hashes[0][:16] if (not is_root and parent_hashes) else None

    # Status from compliance_flags
    flags = artifact.get("compliance_flags", [])
    statuses = {f["status"] for f in flags}
    if "FAIL" in statuses:   status_code = 2
    elif statuses == {"PASS"}: status_code = 1
    else:                      status_code = 0

    start_ns = iso_to_unix_ns(artifact["generated_at"])
    end_ns   = start_ns + duration_ms * 1_000_000

    mandate = artifact.get("mandate_type", "unknown")
    tool_id = artifact.get("tool_id", "unknown")

    attrs = [
        {"key": "ocg.tool_id",       "value": {"stringValue": tool_id}},
        {"key": "ocg.tool_version",   "value": {"stringValue": artifact.get("tool_version", "")}},
        {"key": "ocg.mandate_type",   "value": {"stringValue": mandate}},
        {"key": "ocg.execution_hash", "value": {"stringValue": execution_hash}},
        {"key": "ocg.generated_at",   "value": {"stringValue": artifact.get("generated_at", "")}},
    ]
    sig_block = (artifact.get("audit_signature") or {}).get("signatures", [])
    if sig_block:
        attrs.append({"key": "ocg.signer_keyid", "value": {"stringValue": sig_block[0].get("keyid", "")}})
    if flags:
        attrs.append({"key": "ocg.compliance_flags", "value": {"stringValue": json.dumps(flags)}})

    span = {
        "traceId":           trace_id,
        "spanId":            span_id,
        "name":              f"{mandate}/{tool_id}",
        "kind":              3,
        "startTimeUnixNano": str(start_ns),
        "endTimeUnixNano":   str(end_ns),
        "status":            {"code": status_code},
        "attributes":        attrs,
    }
    if parent_id:
        span["parentSpanId"] = parent_id
    return span

def export_chain_to_otel(artifacts: list, collector_url: str,
                          service_name: str = "ocg-chain",
                          duration_ms: int = 1,
                          headers: dict = None) -> dict:
    """Convert a list of OCG artifacts and POST to an OTel Collector."""
    # Auto-detect root
    all_parents = {h for a in artifacts for h in a.get("parent_hashes", [])}
    root = next((a for a in artifacts if a["execution_hash"] not in all_parents), artifacts[0])
    root_hash = root["execution_hash"]

    spans = [artifact_to_otel_span(a, root_hash, duration_ms) for a in artifacts]
    payload = {
        "resourceSpans": [{
            "resource": {"attributes": [
                {"key": "service.name",     "value": {"stringValue": service_name}},
                {"key": "ocg.root_hash",    "value": {"stringValue": root_hash}},
                {"key": "ocg.chain_length", "value": {"intValue": len(artifacts)}},
            ]},
            "scopeSpans": [{
                "scope": {"name": "ocg-otel-adapter", "version": "0.2.0"},
                "spans": spans
            }]
        }]
    }
    url = collector_url.rstrip("/") + "/v1/traces"
    resp = requests.post(url, json=payload, headers=headers or {}, timeout=10)
    return {"ok": resp.ok, "status": resp.status_code, "body": resp.text}

# Usage:
# result = export_chain_to_otel(artifacts, "http://localhost:4318", service_name="payment-auth")
# assert result["ok"], result["body"]

{
  "mcp_tool_definition": {
    "name": "emit_ocg_otel_trace",
    "description": "Convert a completed OCG v0.3 decision chain (array of artifacts) to OpenTelemetry spans and emit them to the configured OTel Collector. Returns the traceId and spanIds for downstream reference. Call this after a multi-tool OCG chain completes to make the full decision flow visible in Datadog, Jaeger, or Grafana Tempo.",
    "inputSchema": {
      "type": "object",
      "required": ["artifacts"],
      "properties": {
        "artifacts": {
          "type": "array",
          "description": "Array of completed OCG v0.3 artifact objects. Order does not matter — root is auto-detected.",
          "items": { "type": "object" },
          "minItems": 1
        },
        "service_name": {
          "type": "string",
          "description": "OTel service.name resource attribute. Defaults to 'ocg-chain'.",
          "default": "ocg-chain"
        },
        "duration_ms": {
          "type": "integer",
          "description": "Per-span duration in milliseconds. Defaults to 1.",
          "default": 1
        }
      }
    }
  }
}

# Add to the processors section of otel-collector-config.yaml
processors:
  tail_sampling:
    decision_wait: "10s"
    num_traces:    50000
    policies:
      # Always retain chains with any FAIL compliance flag
      - name: retain-fail-chains
        type: string_attribute
        string_attribute:
          key:    "ocg.compliance_flags"
          values: ["*FAIL*"]

      # Always retain capital adequacy assessment chains
      - name: retain-capital-chains
        type: string_attribute
        string_attribute:
          key:    "ocg.mandate_type"
          values: ["capital_assessment"]

      # Sample 10% of routine liquidity_mandate chains
      - name: sample-liquidity-chains
        type: probabilistic
        probabilistic:
          sampling_percentage: 10

      # Drop all remaining traces (adjust as needed)
      - name: drop-rest
        type: always_off