feat(rivetkit): actor tracing, invocation metrics, and log correlation with otel

.claude/reference/testing.md

@@ -47,6 +47,7 @@ For RivetKit runtime or parity bugs, use `rivetkit-typescript/packages/rivetkit`
 
 - Keep RivetKit test fixtures scoped to the engine-only runtime.
 - Prefer targeted integration tests under `rivetkit-typescript/packages/rivetkit/tests/` over shared multi-driver matrices.
+- A span and its parent can arrive in different OTLP export batches, so a trace test that waits for the child and then asserts its `parentSpanId` is racy. Wait on a predicate over the whole exported span list until both are present, then assert the relationship.
 
 ## Frontend testing
 

CLAUDE.md

@@ -392,6 +392,7 @@ Load these only when the task touches the topic.
 - **[SQLite VFS parity](docs-internal/engine/sqlite-vfs.md)** — native Rust VFS ↔ WASM TypeScript VFS 1:1 parity rule, v2 storage keys, chunk layout, delete/truncate strategy. Read before touching either VFS.
 - **[SQLite optimizations](docs-internal/engine/SQLITE_OPTIMIZATIONS.md)** — brief tracker for SQLite cold-read, VFS, storage, preload, and benchmark optimization ideas.
 - **[TLS trust roots](docs-internal/engine/tls-trust-roots.md)** — rustls native+webpki union rationale, which clients use which backend.
+- **[RivetKit telemetry](docs-internal/engine/rivetkit-telemetry.md)** — Core-owned invocation and SQLite spans, ray semantics, schedule trace origins in `_rivet_meta`, native OTLP export. Read before touching actor tracing, metrics, or log correlation.
 - **[Sleep sequence](docs-internal/engine/sleep-sequence.md)** — engine lifecycle authority, `keepAwake` vs `waitUntil` semantics, grace deadline shutdown-token abort, `can_arm_sleep_timer` vs `can_finalize_sleep` predicates. Read before touching sleep/destroy lifecycle.
 
 ### Agent procedural (`.claude/reference/`)

rivetkit-rust/engine/artifacts/errors/actor.operation_abandoned.json

@@ -0,0 +1,5 @@
+{
+  "code": "operation_abandoned",
+  "group": "actor",
+  "message": "Operation tracking ended before a result was recorded."
+}

rivetkit-rust/packages/actor-persist/src/versioned.rs

@@ -1,4 +1,5 @@
 use anyhow::{Result, bail};
+use serde::{Deserialize, Serialize};
 use vbare::OwnedVersionedData;
 
 use crate::generated::{v1, v2, v3, v4};
@@ -500,6 +501,103 @@ pub enum RunWakeAt {
 	V1(Option<i64>),
 }
 
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ScheduleTraceContextData {
+	pub ray_id: Option<String>,
+	pub traceparent: Option<String>,
+	pub tracestate: Option<String>,
+}
+
+pub enum ScheduleTraceContext {
+	V1(ScheduleTraceContextData),
+}
+
+impl OwnedVersionedData for ScheduleTraceContext {
+	type Latest = ScheduleTraceContextData;
+
+	fn wrap_latest(latest: Self::Latest) -> Self {
+		Self::V1(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			Self::V1(data) => Ok(data),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid schedule trace context version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, version: u16) -> Result<Vec<u8>> {
+		match (self, version) {
+			(Self::V1(data), 1) => serde_bare::to_vec(&data).map_err(Into::into),
+			(_, version) => bail!("unexpected schedule trace context version: {version}"),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+}
+
+/// Trace origin of a queue message, stored beside its `_rivet_queue` row. It
+/// has its own versioned type so its persisted meaning cannot be conflated with
+/// `ScheduleTraceContext` even though both currently encode the same fields.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct QueueTraceContextData {
+	pub ray_id: Option<String>,
+	pub traceparent: Option<String>,
+	pub tracestate: Option<String>,
+}
+
+pub enum QueueTraceContext {
+	V1(QueueTraceContextData),
+}
+
+impl OwnedVersionedData for QueueTraceContext {
+	type Latest = QueueTraceContextData;
+
+	fn wrap_latest(latest: Self::Latest) -> Self {
+		Self::V1(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			Self::V1(data) => Ok(data),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid queue trace context version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, version: u16) -> Result<Vec<u8>> {
+		match (self, version) {
+			(Self::V1(data), 1) => serde_bare::to_vec(&data).map_err(Into::into),
+			(_, version) => bail!("unexpected queue trace context version: {version}"),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+}
+
 impl OwnedVersionedData for RunWakeAt {
 	type Latest = Option<i64>;
 

rivetkit-rust/packages/client-protocol/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod generated;
+pub mod telemetry_headers;
 pub mod versioned;
 
 // Re-export latest.

rivetkit-rust/packages/client-protocol/src/telemetry_headers.rs

@@ -0,0 +1,37 @@
+//! Headers that carry a caller's ray and W3C trace context into an actor,
+//! shared by every Rust peer of the actor HTTP surface: the runtime that reads
+//! them and the clients that send them.
+
+/// Bounded correlation string that follows work across actors, surviving
+/// sampling and trace-root boundaries.
+pub const HEADER_RIVETKIT_RAY_ID: &str = "x-rivetkit-ray-id";
+pub const HEADER_TRACEPARENT: &str = "traceparent";
+pub const HEADER_TRACESTATE: &str = "tracestate";
+
+/// W3C Baggage key that carries a ray through application code, so a request
+/// handler that received a ray can hand it to the actors it calls.
+pub const RAY_BAGGAGE_KEY: &str = "rivetkit.ray";
+
+const RAY_ID_MAX_LEN: usize = 128;
+
+/// Returns `value` when it is a ray the runtime accepts: 1 to 128 characters
+/// of `[A-Za-z0-9_-]`. The value arrives from a caller, so anything else
+/// counts as absent, and the receiving invocation mints a fresh ray instead.
+pub fn bounded_ray_id(value: &str) -> Option<&str> {
+	let valid = !value.is_empty()
+		&& value.len() <= RAY_ID_MAX_LEN
+		&& value
+			.bytes()
+			.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'));
+	if valid { Some(value) } else { None }
+}
+
+/// Formats a version 00 W3C `traceparent` from the parts of a span context,
+/// so every Rust peer writes the header the same way.
+pub fn format_traceparent(
+	trace_id: impl std::fmt::Display,
+	span_id: impl std::fmt::Display,
+	trace_flags: u8,
+) -> String {
+	format!("00-{trace_id}-{span_id}-{trace_flags:02x}")
+}

rivetkit-rust/packages/client/Cargo.toml

@@ -13,6 +13,7 @@ anyhow = "1.0"
 base64 = "0.22.1"
 bytes = { workspace = true }
 futures-util = "0.3.31"
+opentelemetry = { version = "0.28", default-features = false, features = ["trace"] }
 parking_lot.workspace = true
 reqwest = { version = "0.12.12", default-features = false, features = ["json", "charset", "http2", "macos-system-configuration", "rustls-tls-native-roots", "rustls-tls-webpki-roots"] }
 rivetkit-client-protocol.workspace = true
@@ -24,12 +25,14 @@ serde_json = "1.0"
 tokio =  { version = "1", features = ["full"] }
 tokio-tungstenite = { version = "0.26.1", features = ["rustls-tls-native-roots", "rustls-tls-webpki-roots", "handshake"] }
 tracing = "0.1.41"
+tracing-opentelemetry = { version = "0.29", default-features = false }
 tungstenite = "0.26.2"
 urlencoding = "2.1.3"
 vbare = "0.0.4"
 
 [dev-dependencies]
 axum = { workspace = true, features = ["ws"] }
+opentelemetry_sdk = { version = "0.28", default-features = false, features = ["trace"] }
 tracing-subscriber = { version = "0.3.19", features = ["env-filter", "std", "registry"]}
 tempfile = "3.10.1"
 tokio-test = "0.4.3"