diff --git a/.github/workflows/build-development-artifacts.yml b/.github/workflows/build-development-artifacts.yml new file mode 100644 index 00000000..2f09afaf --- /dev/null +++ b/.github/workflows/build-development-artifacts.yml @@ -0,0 +1,89 @@ +name: Build OpenAPPA Development Artifacts + +on: + pull_request: + types: + - labeled + - synchronize + +permissions: {} + +concurrency: + group: "openappa-development-${{ github.event.pull_request.number }}" + cancel-in-progress: true + +jobs: + build: + name: Build Linux development artifacts + if: >- + ${{ github.event.pull_request.head.repo.full_name == github.repository && + ((github.event.action == 'labeled' && github.event.label.name == 'build-development-artifacts') || + (github.event.action == 'synchronize' && contains(github.event.pull_request.labels.*.name, 'build-development-artifacts'))) }} + runs-on: ubuntu-22.04 + timeout-minutes: 30 + permissions: + contents: write # Publish the opt-in development prerelease and its immutable assets. + steps: + - name: Checkout the reviewed PR commit + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + fetch-depth: 1 + + - name: Resolve development release + id: release + env: + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: | + short_sha=${HEAD_SHA:0:12} + release_ref="dev-pr-${PR_NUMBER}-${short_sha}" + echo "release_ref=${release_ref}" >> "$GITHUB_OUTPUT" + echo "version=${release_ref#dev-}" >> "$GITHUB_OUTPUT" + + - name: Stage plugin and build runtime + env: + APPA_RELEASE_REF: ${{ steps.release.outputs.release_ref }} + run: | + mkdir -p dist + sh scripts/appa-stage-plugin-bundle.sh staged + version=$(cargo metadata --locked --no-deps --format-version 1 \ + | jq -r '.packages[] | select(.name == "appa") | .version') + tar -C staged -czf "dist/appa-plugin-${version}.tar.gz" . + cp "dist/appa-plugin-${version}.tar.gz" "dist/appa-plugin-${APPA_RELEASE_REF}.tar.gz" + plugin_sha256=$(sha256sum "dist/appa-plugin-${version}.tar.gz" | cut -d' ' -f1) + APPA_PLUGIN_SHA256="$plugin_sha256" cargo build --release --locked --package appa + mkdir -p package + cp target/release/appa package/appa + chmod 755 package/appa + tar -C package -czf "dist/appa-x86_64-unknown-linux-gnu.tar.gz" appa + (cd dist && sha256sum *.tar.gz > SHA256SUMS) + + - name: Publish immutable development prerelease + env: + GH_TOKEN: ${{ github.token }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + PR_NUMBER: ${{ github.event.pull_request.number }} + RELEASE_REF: ${{ steps.release.outputs.release_ref }} + run: | + gh release create "$RELEASE_REF" dist/* \ + --repo "$GITHUB_REPOSITORY" \ + --target "$HEAD_SHA" \ + --title "OpenAPPA ${RELEASE_REF}" \ + --notes "Development artifacts for PR #${PR_NUMBER} at ${HEAD_SHA}. Not a supported release." \ + --prerelease + + - name: Publish usage summary + env: + RELEASE_REF: ${{ steps.release.outputs.release_ref }} + run: | + { + echo "### OpenAPPA development build" + echo + echo "Release: https://github.com/${GITHUB_REPOSITORY}/releases/tag/${RELEASE_REF}" + echo + echo '```dockerfile' + echo "ARG APPA_VERSION=${RELEASE_REF}" + echo '```' + } >> "$GITHUB_STEP_SUMMARY" diff --git a/.github/workflows/pr-title-linter.yml b/.github/workflows/pr-title-linter.yml index 9748f718..aa53c4cd 100644 --- a/.github/workflows/pr-title-linter.yml +++ b/.github/workflows/pr-title-linter.yml @@ -36,7 +36,7 @@ jobs: - name: Lint PR title if: ${{ github.event_name == 'pull_request' }} - uses: grafana/shared-workflows/actions/lint-pr-title@70c40e5b6522c854a55334124545261cae3d4c73 # v1.2.4 + uses: grafana/shared-workflows/actions/lint-pr-title@70c40e5b6522c854a55334124545261cae3d4c73 # lint-pr-title/v1.2.4 with: config-path: "${{ github.workspace }}/.github/commitlint.config.js" env: diff --git a/Cargo.lock b/Cargo.lock index 7d231129..850d7813 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -120,6 +120,10 @@ dependencies = [ "jsonschema", "libc", "libloading", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry_sdk", "rand 0.10.2", "reqwest", "rig-agent", @@ -136,6 +140,7 @@ dependencies = [ "tokio", "toml", "tracing", + "tracing-opentelemetry", "tracing-subscriber", "url", ] @@ -722,6 +727,12 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + [[package]] name = "email_address" version = "0.2.9" @@ -1325,6 +1336,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1689,6 +1709,89 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry-appender-tracing" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837" +dependencies = [ + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", +] + +[[package]] +name = "opentelemetry-http" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest", + "thiserror", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.5", + "thiserror", + "tokio", +] + [[package]] name = "ordered-float" version = "5.5.0" @@ -1823,6 +1926,29 @@ dependencies = [ "unarray", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "pyo3" version = "0.29.0" @@ -2046,6 +2172,7 @@ dependencies = [ "base64 0.22.1", "bytes", "encoding_rs", + "futures-channel", "futures-core", "futures-util", "h2", @@ -3014,6 +3141,20 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" +dependencies = [ + "js-sys", + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -3264,6 +3405,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-root-certs" version = "1.0.9" diff --git a/appa-runtime/Cargo.toml b/appa-runtime/Cargo.toml index 9a2c52a5..3376ab16 100644 --- a/appa-runtime/Cargo.toml +++ b/appa-runtime/Cargo.toml @@ -30,6 +30,10 @@ futures-util = { version = "0.3", default-features = false, features = ["alloc"] libloading = "0.9" libc = "0.2" flate2 = { version = "1", default-features = false, features = ["rust_backend"] } +opentelemetry = { version = "0.32", features = ["logs", "metrics", "trace"] } +opentelemetry-appender-tracing = "0.32" +opentelemetry-otlp = { version = "0.32", default-features = false, features = ["http-proto", "logs", "metrics", "reqwest-blocking-client", "trace"] } +opentelemetry_sdk = { version = "0.32", features = ["logs", "metrics", "trace"] } rand = "0.10" reqwest = { workspace = true } # `server` carries the tool macros' schema derive; `transport-streamable-http-server` @@ -43,9 +47,10 @@ sha2 = { workspace = true } thiserror = { workspace = true } tar = "0.4" tempfile = "3" -tokio = { workspace = true } +tokio = { workspace = true, features = ["signal"] } toml = { workspace = true } tracing = { workspace = true } +tracing-opentelemetry = { version = "0.33", default-features = false } tracing-subscriber = { version = "0.3", features = ["env-filter"] } url = "2" # Not rig-core's `rustls` feature: it pins reqwest to aws-lc-rs, which would put a @@ -63,6 +68,9 @@ tar = "0.4" # The fail points the orchestration tests drive. A dev-dependency, so they are # absent from the binary. appa-eventlog = { workspace = true, features = ["fault-injection"] } +# The in-memory span exporter verifies the semantic attributes that leave the +# tracing bridge, rather than only the runtime's intermediate projection. +opentelemetry_sdk = { version = "0.32", features = ["testing", "trace"] } jsonschema = { version = "0.53.0", default-features = false } # The elicitation round trip is only real with a real client on the # other end: these features exist for `tests/hitl_elicitation.rs` and diff --git a/appa-runtime/src/hooks.rs b/appa-runtime/src/hooks.rs index e9fa3fc9..408d18b8 100644 --- a/appa-runtime/src/hooks.rs +++ b/appa-runtime/src/hooks.rs @@ -1,7 +1,12 @@ //! The hook dispatcher: one typed `HookEvent` in, one `HookDecision` //! out. +use std::time::Instant; + use appa_runtime_api::{Actor, Codec, HookDecision, HookEvent, ParseRefusal, ProposedCall, Ruling, TrajectoryId}; +use sha2::{Digest, Sha256}; +use tracing::Instrument; +use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::api::{ ChildReturnDecision, EventError, LateOpen, OfferId, Runtime, Session, SpawnResultDecision, ToolCallDecision, @@ -13,28 +18,211 @@ use crate::api::{ /// non-2xx status makes the hook command exit 2, which blocks the /// action — hooks fail closed. pub async fn answer(runtime: &Runtime, codec: &Codec, body: &[u8]) -> (u16, serde_json::Value) { + let started = Instant::now(); let event = match (codec.parse)(body) { Ok(Some(event)) => event, - Ok(None) => return (200, serde_json::json!({})), - Err(ParseRefusal::Unreadable { detail }) => return (400, serde_json::json!({ "error": detail })), - Err(ParseRefusal::Malformed { detail }) => return (409, serde_json::json!({ "error": detail })), + Ok(None) => { + crate::telemetry::record_hook("ignored", "ack", started.elapsed().as_secs_f64()); + return (200, serde_json::json!({})); + } + Err(ParseRefusal::Unreadable { detail }) => { + observe_parse_refusal("unreadable", started.elapsed().as_secs_f64()); + return (400, serde_json::json!({ "error": detail })); + } + Err(ParseRefusal::Malformed { detail }) => { + observe_parse_refusal("malformed", started.elapsed().as_secs_f64()); + return (409, serde_json::json!({ "error": detail })); + } }; - if let HookEvent::ToolCall { actor, call, .. } = &event { - match runtime.opened_among(&actor.root, &(codec.names_children)(actor, call)) { - Ok(Some(child)) => { - tracing::debug!(root = %actor.root.0, child = %child.0, "a call names a family child's transcript"); - return (200, (codec.render)(&event, &deny(NAMED_TRANSCRIPT.to_string()))); + let observation = HookObservation::from_event(&event, ToolArgumentCapture::from_env()); + let span = hook_span(&observation); + async move { + let decision = match &event { + HookEvent::ToolCall { actor, call, .. } => { + match runtime.opened_among(&actor.root, &(codec.names_children)(actor, call)) { + Ok(Some(child)) => { + tracing::debug!( + root = %actor.root.0, + child = %child.0, + "a call names a family child's transcript" + ); + deny(NAMED_TRANSCRIPT.to_string()) + } + Ok(None) => handle(runtime, event.clone()).await, + Err(error) => refuse(error.to_string()), + } } - Ok(None) => {} - Err(error) => return (409, (codec.render)(&event, &refuse(error.to_string()))), + _ => handle(runtime, event.clone()).await, + }; + let decision_name = decision_name(&decision); + tracing::Span::current().record("appa.decision", decision_name); + if matches!(decision, HookDecision::Refuse { .. }) { + tracing::Span::current().record("error.type", "appa.runtime.refusal"); + tracing::Span::current().record("otel.status_code", "ERROR"); } + tracing::info!( + "appa.hook.event" = observation.event, + "appa.trajectory.id" = %observation.root, + "appa.trajectory.child_id" = observation.child.as_deref().unwrap_or(""), + "gen_ai.conversation.id" = %observation.root, + "gen_ai.tool.name" = observation.tool.as_deref().unwrap_or(""), + "appa.decision" = decision_name, + "hook decision" + ); + crate::telemetry::record_hook(observation.event, decision_name, started.elapsed().as_secs_f64()); + let status = match decision { + HookDecision::Refuse { .. } => 409, + _ => 200, + }; + (status, (codec.render)(&event, &decision)) } - let decision = handle(runtime, event.clone()).await; - let status = match decision { - HookDecision::Refuse { .. } => 409, - _ => 200, - }; - (status, (codec.render)(&event, &decision)) + .instrument(span) + .await +} + +fn hook_span(observation: &HookObservation) -> tracing::Span { + let span = tracing::info_span!( + "appa.hook", + "appa.hook.event" = observation.event, + "appa.trajectory.id" = %observation.root, + "appa.trajectory.child_id" = observation.child.as_deref().unwrap_or(""), + "gen_ai.conversation.id" = %observation.root, + "gen_ai.tool.name" = observation.tool.as_deref().unwrap_or(""), + "appa.decision" = tracing::field::Empty, + "error.type" = tracing::field::Empty, + "otel.status_code" = tracing::field::Empty, + ); + match &observation.arguments { + ToolArguments::NotCaptured => {} + ToolArguments::Captured(arguments) => { + span.set_attribute("gen_ai.tool.call.arguments", arguments.clone()); + } + ToolArguments::Omitted { size_bytes, sha256 } => { + span.set_attribute( + "appa.tool.call.arguments.size_bytes", + i64::try_from(*size_bytes).unwrap_or(i64::MAX), + ); + span.set_attribute("appa.tool.call.arguments.sha256", sha256.clone()); + } + } + span +} + +struct HookObservation { + event: &'static str, + root: String, + child: Option, + tool: Option, + arguments: ToolArguments, +} + +impl HookObservation { + fn from_event(event: &HookEvent, capture: ToolArgumentCapture) -> Self { + match event { + HookEvent::SessionStart { root } => Self::new("session_start", root, None, None, capture), + HookEvent::Prompt { actor, .. } => Self::actor("prompt", actor, None, capture), + HookEvent::TurnEnd { actor } => Self::actor("turn_end", actor, None, capture), + HookEvent::ToolCall { actor, call, .. } => Self::actor("tool_call", actor, Some(call), capture), + HookEvent::ToolResult { actor, call, .. } => Self::actor("tool_result", actor, Some(call), capture), + HookEvent::ChildStart { root, child, .. } => Self::new("child_start", root, Some(child), None, capture), + HookEvent::ChildEnd { root, child, .. } => Self::new("child_end", root, Some(child), None, capture), + HookEvent::SpawnResult { actor, call, .. } => Self::actor("spawn_result", actor, Some(call), capture), + } + } + + fn actor(event: &'static str, actor: &Actor, call: Option<&ProposedCall>, capture: ToolArgumentCapture) -> Self { + Self::new(event, &actor.root, actor.child.as_ref(), call, capture) + } + + fn new( + event: &'static str, + root: &TrajectoryId, + child: Option<&TrajectoryId>, + call: Option<&ProposedCall>, + capture: ToolArgumentCapture, + ) -> Self { + Self { + event, + root: root.0.clone(), + child: child.map(|id| id.0.clone()), + tool: call.map(|call| call.tool.clone()), + arguments: ToolArguments::from_call(call, capture), + } + } +} + +#[derive(Clone, Copy)] +enum ToolArgumentCapture { + Disabled, + Enabled, +} + +impl ToolArgumentCapture { + fn from_env() -> Self { + let value = std::env::var(CAPTURE_TOOL_ARGUMENTS_ENV).ok(); + Self::from_value(value.as_deref()) + } + + fn from_value(value: Option<&str>) -> Self { + match value { + Some(value) if value.eq_ignore_ascii_case("true") => Self::Enabled, + _ => Self::Disabled, + } + } +} + +#[derive(Debug, PartialEq, Eq)] +enum ToolArguments { + NotCaptured, + Captured(String), + Omitted { size_bytes: usize, sha256: String }, +} + +impl ToolArguments { + fn from_call(call: Option<&ProposedCall>, capture: ToolArgumentCapture) -> Self { + let (Some(call), ToolArgumentCapture::Enabled) = (call, capture) else { + return Self::NotCaptured; + }; + let arguments = call.arguments.get(); + if arguments.len() <= MAX_CAPTURED_TOOL_ARGUMENT_BYTES { + return Self::Captured(arguments.to_owned()); + } + Self::Omitted { + size_bytes: arguments.len(), + sha256: format!("{:x}", Sha256::digest(arguments.as_bytes())), + } + } +} + +const CAPTURE_TOOL_ARGUMENTS_ENV: &str = "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"; +const MAX_CAPTURED_TOOL_ARGUMENT_BYTES: usize = 32 * 1024; + +fn decision_name(decision: &HookDecision) -> &'static str { + match decision { + HookDecision::Ack => "ack", + HookDecision::AllowCall { .. } => "allow", + HookDecision::PassControl => "pass_control", + HookDecision::DenyCall { .. } => "deny", + HookDecision::Block { .. } => "block", + HookDecision::ReplaceOutput { .. } => "replace_output", + HookDecision::ChildReturn { .. } => "replace_child_return", + HookDecision::Context { .. } => "context", + HookDecision::Refuse { .. } => "refuse", + } +} + +fn observe_parse_refusal(kind: &'static str, elapsed_seconds: f64) { + let span = tracing::info_span!( + "appa.hook", + "appa.hook.event" = "parse", + "appa.decision" = "refuse", + "error.type" = kind, + "otel.status_code" = "ERROR", + ); + span.in_scope(|| { + tracing::warn!("error.type" = kind, "hook input refused"); + }); + crate::telemetry::record_hook("parse", "refuse", elapsed_seconds); } /// A subagent's words reach its parent through the checked return only; a call that @@ -380,6 +568,109 @@ mod tests { Config::load(&path).expect("the minimal fixture validates") } + #[test] + fn hook_observation_does_not_capture_call_data_without_opt_in() { + let event = HookEvent::ToolCall { + actor: Actor { + root: TrajectoryId("cc:root".to_string()), + child: Some(TrajectoryId("cc:child".to_string())), + }, + call: ProposedCall { + tool: "Write".to_string(), + arguments: serde_json::value::RawValue::from_string(r#"{"secret":"not telemetry"}"#.to_string()) + .expect("valid arguments"), + }, + spawn: false, + ruling: None, + }; + + let observation = HookObservation::from_event(&event, ToolArgumentCapture::Disabled); + assert_eq!(observation.event, "tool_call"); + assert_eq!(observation.root, "cc:root"); + assert_eq!(observation.child.as_deref(), Some("cc:child")); + assert_eq!(observation.tool.as_deref(), Some("Write")); + assert_eq!(observation.arguments, ToolArguments::NotCaptured); + } + + #[test] + fn opted_in_tool_arguments_keep_the_harness_json_spelling() { + let event = tool_call("cc:root", r#"{"a":1,"a":2}"#); + + let observation = HookObservation::from_event(&event, ToolArgumentCapture::Enabled); + assert_eq!( + observation.arguments, + ToolArguments::Captured(r#"{"a":1,"a":2}"#.to_string()) + ); + } + + #[test] + fn oversized_tool_arguments_emit_only_size_and_digest_metadata() { + let arguments = format!(r#"{{"body":"{}"}}"#, "x".repeat(MAX_CAPTURED_TOOL_ARGUMENT_BYTES)); + let event = HookEvent::ToolCall { + actor: Actor { + root: TrajectoryId("cc:root".to_string()), + child: None, + }, + call: ProposedCall { + tool: "Write".to_string(), + arguments: serde_json::value::RawValue::from_string(arguments.clone()).expect("valid arguments"), + }, + spawn: false, + ruling: None, + }; + + let observation = HookObservation::from_event(&event, ToolArgumentCapture::Enabled); + let ToolArguments::Omitted { size_bytes, sha256 } = observation.arguments else { + panic!("oversized arguments must be omitted"); + }; + assert_eq!(size_bytes, arguments.len()); + assert_eq!( + sha256, + "411932b4696bf0016b1e8e068dd2f32037bab015a0b2ba26df0117b8c7b2471f" + ); + } + + #[test] + fn tool_argument_capture_requires_an_explicit_true_value() { + assert!(matches!( + ToolArgumentCapture::from_value(Some("true")), + ToolArgumentCapture::Enabled + )); + assert!(matches!( + ToolArgumentCapture::from_value(Some("TRUE")), + ToolArgumentCapture::Enabled + )); + for value in [None, Some(""), Some("1"), Some("false"), Some("yes")] { + assert!(matches!( + ToolArgumentCapture::from_value(value), + ToolArgumentCapture::Disabled + )); + } + } + + fn tool_call(root: &str, arguments: &str) -> HookEvent { + HookEvent::ToolCall { + actor: Actor { + root: TrajectoryId(root.to_string()), + child: None, + }, + call: ProposedCall { + tool: "Bash".to_string(), + arguments: serde_json::value::RawValue::from_string(arguments.to_string()).expect("valid arguments"), + }, + spawn: false, + ruling: None, + } + } + + #[test] + fn hook_decisions_have_bounded_telemetry_names() { + assert_eq!(decision_name(&HookDecision::Ack), "ack"); + assert_eq!(decision_name(&deny("no".to_string())), "deny"); + assert_eq!(decision_name(&block("no".to_string())), "block"); + assert_eq!(decision_name(&refuse("no".to_string())), "refuse"); + } + fn open_runtime(dir: &tempfile::TempDir) -> Runtime { Runtime::open(config(), dir.path().join("appa.db"), None).expect("the fixture deployment opens") } diff --git a/appa-runtime/src/lib.rs b/appa-runtime/src/lib.rs index 03fe7e54..682d9b2c 100644 --- a/appa-runtime/src/lib.rs +++ b/appa-runtime/src/lib.rs @@ -13,6 +13,7 @@ pub mod plugin_bundle; pub mod replay; #[path = "main.rs"] pub mod runtime_cli; +pub(crate) mod telemetry; pub mod tls; mod builtins; diff --git a/appa-runtime/src/main.rs b/appa-runtime/src/main.rs index ee668d37..a8bf7152 100644 --- a/appa-runtime/src/main.rs +++ b/appa-runtime/src/main.rs @@ -301,12 +301,10 @@ where } async fn serve(args: Args) -> ExitCode { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(log_level(args.verbose))), - ) - .init(); + let level = tracing_subscriber::EnvFilter::try_from_default_env() + .map(|filter| filter.to_string()) + .unwrap_or_else(|_| log_level(args.verbose).to_string()); + let telemetry = crate::telemetry::Telemetry::init(&level); let config_path = args.config.unwrap_or_else(|| PathBuf::from("appa.toml")); @@ -377,12 +375,36 @@ async fn serve(args: Args) -> ExitCode { listen = %args.listen, "appa-runtime serving /hook, /mcp, /health, and /batteries; management routes require loopback" ); - match axum::serve(listener, app.into_make_service_with_connect_info::()).await { + let exit = match axum::serve(listener, app.into_make_service_with_connect_info::()) + .with_graceful_shutdown(shutdown_signal()) + .await + { Ok(()) => ExitCode::SUCCESS, Err(error) => { eprintln!("appa runtime: server failed: {error}"); ExitCode::FAILURE } + }; + telemetry.shutdown(); + exit +} + +async fn shutdown_signal() { + #[cfg(unix)] + { + let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("SIGTERM handling is available"); + tokio::select! { + result = tokio::signal::ctrl_c() => { + let _ = result; + } + _ = terminate.recv() => {} + } + } + + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; } } diff --git a/appa-runtime/src/telemetry.rs b/appa-runtime/src/telemetry.rs new file mode 100644 index 00000000..cc916259 --- /dev/null +++ b/appa-runtime/src/telemetry.rs @@ -0,0 +1,228 @@ +//! Operational telemetry for the runtime process. +//! +//! The engine stays unaware of exporters. This module projects the runtime's +//! existing `tracing` spans and events to OTLP when an OTLP endpoint is +//! configured, and leaves the ordinary stderr subscriber unchanged otherwise. + +use std::sync::LazyLock; +use std::time::Instant; + +use opentelemetry::metrics::{Counter, Histogram}; +use opentelemetry::{KeyValue, global}; +use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::logs::SdkLoggerProvider; +use opentelemetry_sdk::metrics::{Aggregation, Instrument, SdkMeterProvider, Stream}; +use opentelemetry_sdk::trace::SdkTracerProvider; +use tracing_subscriber::filter::Filtered; +use tracing_subscriber::fmt::Layer as FmtLayer; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::{EnvFilter, Layer, Registry, fmt, util::SubscriberInitExt}; + +/// Providers kept alive for as long as the runtime serves. Dropping an SDK +/// provider does not flush it, so `shutdown` is explicit on an orderly exit. +pub(crate) struct Telemetry { + tracer: Option, + logger: Option, + meter: Option, +} + +impl Telemetry { + /// Install stderr logging and, when an OTLP endpoint is configured, the + /// non-blocking trace, log, and metric exporters. + pub(crate) fn init(level: &str) -> Self { + if !otlp_configured() { + tracing_subscriber::registry().with(stderr_layer(level)).init(); + return Self::disabled(); + } + + match Self::init_otlp(level) { + Ok(telemetry) => telemetry, + Err(error) => { + eprintln!("appa runtime: OTLP telemetry is disabled: {error}"); + tracing_subscriber::registry().with(stderr_layer(level)).init(); + Self::disabled() + } + } + } + + /// Flush all three signals without allowing exporter failure to change the + /// runtime's exit status. + pub(crate) fn shutdown(self) { + if let Some(provider) = self.meter { + let _ = provider.shutdown(); + } + if let Some(provider) = self.logger { + let _ = provider.shutdown(); + } + if let Some(provider) = self.tracer { + let _ = provider.shutdown(); + } + } + + fn init_otlp(level: &str) -> Result> { + use opentelemetry::trace::TracerProvider as _; + + crate::tls::install_crypto_provider(); + let resource = Resource::builder() + .with_service_name("appa-runtime") + .with_attribute(KeyValue::new("service.version", env!("CARGO_PKG_VERSION"))) + .build(); + + let span_exporter = opentelemetry_otlp::SpanExporter::builder().with_http().build()?; + let log_exporter = opentelemetry_otlp::LogExporter::builder().with_http().build()?; + let metric_exporter = opentelemetry_otlp::MetricExporter::builder().with_http().build()?; + + let tracer = SdkTracerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(span_exporter) + .build(); + let logger = SdkLoggerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(log_exporter) + .build(); + let meter = SdkMeterProvider::builder() + .with_resource(resource) + .with_periodic_exporter(metric_exporter) + .with_view(hook_duration_view) + .build(); + + global::set_meter_provider(meter.clone()); + register_runtime_uptime(); + let trace_layer = tracing_opentelemetry::layer() + .with_tracer(tracer.tracer("appa-runtime")) + .with_filter(otel_filter(level)); + let log_layer = OpenTelemetryTracingBridge::new(&logger).with_filter(otel_filter(level)); + + let subscriber_result = tracing_subscriber::registry() + .with(stderr_layer(level)) + .with(trace_layer) + .with(log_layer) + .try_init(); + + if let Err(error) = subscriber_result { + let _ = meter.shutdown(); + let _ = logger.shutdown(); + let _ = tracer.shutdown(); + return Err(Box::new(error)); + } + + Ok(Self { + tracer: Some(tracer), + logger: Some(logger), + meter: Some(meter), + }) + } + + fn disabled() -> Self { + Self { + tracer: None, + logger: None, + meter: None, + } + } +} + +/// Record only bounded labels. Trajectory and tool identity belong on spans +/// and logs, never metric dimensions. +pub(crate) fn record_hook(event: &'static str, decision: &'static str, elapsed_seconds: f64) { + let attributes = [ + KeyValue::new("appa.hook.event", event), + KeyValue::new("appa.decision", decision), + ]; + HOOKS.add(1, &attributes); + HOOK_DURATION.record(elapsed_seconds, &attributes); +} + +fn otlp_configured() -> bool { + otlp_configured_with(|name| std::env::var_os(name)) +} + +fn otlp_configured_with(mut value: impl FnMut(&str) -> Option) -> bool { + [ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + ] + .into_iter() + .any(|name| value(name).is_some_and(|value| !value.is_empty())) +} + +fn stderr_layer(level: &str) -> Filtered, EnvFilter, Registry> { + fmt::layer().with_filter(runtime_filter(level)) +} + +fn otel_filter(level: &str) -> EnvFilter { + runtime_filter(level) +} + +fn runtime_filter(level: &str) -> EnvFilter { + EnvFilter::new(level) + .add_directive("opentelemetry=off".parse().expect("the OpenTelemetry filter is valid")) + .add_directive( + "opentelemetry_sdk=off" + .parse() + .expect("the OpenTelemetry SDK filter is valid"), + ) + .add_directive("opentelemetry_otlp=off".parse().expect("the OTLP filter is valid")) + .add_directive("hyper=off".parse().expect("the Hyper filter is valid")) + .add_directive("hyper_util=off".parse().expect("the Hyper utility filter is valid")) + .add_directive("reqwest=off".parse().expect("the reqwest filter is valid")) +} + +fn hook_duration_view(instrument: &Instrument) -> Option { + if instrument.name() != "appa.runtime.hook.duration" { + return None; + } + Some( + Stream::builder() + .with_aggregation(Aggregation::ExplicitBucketHistogram { + boundaries: vec![ + 0.000_1, 0.000_5, 0.001, 0.002_5, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, + ], + record_min_max: true, + }) + .build() + .expect("hook latency histogram view is valid"), + ) +} + +fn register_runtime_uptime() { + let started_at = Instant::now(); + global::meter("appa-runtime") + .u64_observable_gauge("appa.runtime.uptime") + .with_description("Seconds since the OpenAPPA runtime started") + .with_unit("s") + .with_callback(move |observer| observer.observe(started_at.elapsed().as_secs(), &[])) + .build(); +} + +static HOOKS: LazyLock> = LazyLock::new(|| { + global::meter("appa-runtime") + .u64_counter("appa.runtime.hook.requests") + .with_description("Hook decisions made by the OpenAPPA runtime") + .build() +}); + +static HOOK_DURATION: LazyLock> = LazyLock::new(|| { + global::meter("appa-runtime") + .f64_histogram("appa.runtime.hook.duration") + .with_description("Elapsed time to answer an OpenAPPA hook") + .with_unit("s") + .build() +}); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn otlp_requires_a_nonempty_endpoint() { + assert!(!otlp_configured_with(|_| None)); + assert!(otlp_configured_with(|name| { + (name == "OTEL_EXPORTER_OTLP_ENDPOINT").then(|| "http://127.0.0.1:4318".into()) + })); + assert!(!otlp_configured_with(|_| Some("".into()))); + } +} diff --git a/appa-runtime/tests/crash_recovery.rs b/appa-runtime/tests/crash_recovery.rs index 60348820..fc2b9d22 100644 --- a/appa-runtime/tests/crash_recovery.rs +++ b/appa-runtime/tests/crash_recovery.rs @@ -43,7 +43,7 @@ fn free_port() -> u16 { } fn start(config: &Path, db: &Path, port: u16) -> Server { - let child = Command::new(env!("CARGO_BIN_EXE_appa")) + let child = isolated_runtime_command() .arg("runtime") .arg("--config") .arg(config) @@ -105,7 +105,7 @@ fn http(url: &str, method: &str, body: Option<&str>) -> Option { } fn expect_startup_refusal(config: &Path, db: &Path, needle: &str) { - let mut child = Command::new(env!("CARGO_BIN_EXE_appa")) + let mut child = isolated_runtime_command() .arg("runtime") .arg("--config") .arg(config) @@ -144,6 +144,19 @@ fn expect_startup_refusal(config: &Path, db: &Path, needle: &str) { ); } +fn isolated_runtime_command() -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_appa")); + for name in [ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + ] { + command.env_remove(name); + } + command +} + fn write_config(dir: &Path, text: &str) -> PathBuf { let path = dir.join("appa.toml"); std::fs::write(&path, text).expect("the config writes"); diff --git a/appa-runtime/tests/tool_argument_telemetry.rs b/appa-runtime/tests/tool_argument_telemetry.rs new file mode 100644 index 00000000..5e14853f --- /dev/null +++ b/appa-runtime/tests/tool_argument_telemetry.rs @@ -0,0 +1,119 @@ +use std::ffi::OsString; + +use appa_runtime::api::Runtime; +use appa_runtime::config::Config; +use appa_runtime::hooks::answer; +use opentelemetry::trace::TracerProvider as _; +use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; +use opentelemetry_sdk::logs::{InMemoryLogExporter, SdkLoggerProvider}; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; +use tracing_subscriber::layer::SubscriberExt as _; + +#[test] +fn opted_in_arguments_leave_on_spans_but_not_logs() { + let _capture = ScopedCaptureOptIn::new(); + let span_exporter = InMemorySpanExporter::default(); + let tracer_provider = SdkTracerProvider::builder() + .with_simple_exporter(span_exporter.clone()) + .build(); + let log_exporter = InMemoryLogExporter::default(); + let logger_provider = SdkLoggerProvider::builder() + .with_simple_exporter(log_exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(tracer_provider.tracer("appa-runtime-test"))) + .with(OpenTelemetryTracingBridge::new(&logger_provider)); + let _subscriber = tracing::subscriber::set_default(subscriber); + + let dir = tempfile::tempdir().expect("a temp dir is creatable"); + let config_path = dir.path().join("appa.toml"); + std::fs::write( + &config_path, + r#" + [policy] + version = 2 + + [[policy.tool]] + name = "Bash" + + [externals] + timeout_ms = 1000 + max_body_bytes = 4096 + "#, + ) + .expect("the fixture config writes"); + let config = Config::load(&config_path).expect("the fixture config loads"); + let runtime = Runtime::open(config, dir.path().join("appa.db"), None).expect("the fixture runtime opens"); + let codec = appa_adapter_claude_code::codec(); + let tokio = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("the test runtime builds"); + tokio.block_on(async { + let start = br#"{"hook_event_name":"SessionStart","session_id":"telemetry-test"}"#; + assert_eq!(answer(&runtime, &codec, start).await.0, 200); + let call = br#"{"hook_event_name":"PreToolUse","session_id":"telemetry-test","tool_name":"Bash","tool_input":{"command":"git status","secret":"span-only"}}"#; + assert_eq!(answer(&runtime, &codec, call).await.0, 200); + }); + drop(_subscriber); + tracer_provider.force_flush().expect("the spans flush"); + logger_provider.force_flush().expect("the logs flush"); + + let spans = span_exporter + .get_finished_spans() + .expect("the exported spans are readable"); + let tool_call = spans + .iter() + .find(|span| span_attribute(span, "appa.hook.event").as_deref() == Some("tool_call")) + .expect("the tool-call span exported"); + assert_eq!( + span_attribute(tool_call, "gen_ai.tool.call.arguments").as_deref(), + Some(r#"{"command":"git status","secret":"span-only"}"#) + ); + + let logs = log_exporter.get_emitted_logs().expect("the exported logs are readable"); + assert!(logs.iter().any(|log| log.record.trace_context().is_some())); + for log in logs { + assert!( + log.record + .attributes_iter() + .all(|(key, _)| key.as_str() != "gen_ai.tool.call.arguments") + ); + assert!(!format!("{:?}", log.record.body()).contains("span-only")); + } +} + +fn span_attribute(span: &opentelemetry_sdk::trace::SpanData, key: &str) -> Option { + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == key) + .map(|attribute| attribute.value.as_str().into_owned()) +} + +struct ScopedCaptureOptIn { + previous: Option, +} + +impl ScopedCaptureOptIn { + fn new() -> Self { + let previous = std::env::var_os(CAPTURE_TOOL_ARGUMENTS_ENV); + // This integration-test executable contains one synchronous test. It + // changes the variable before it creates a runtime or telemetry task. + unsafe { std::env::set_var(CAPTURE_TOOL_ARGUMENTS_ENV, "true") }; + Self { previous } + } +} + +impl Drop for ScopedCaptureOptIn { + fn drop(&mut self) { + // The test remains single-threaded until this guard restores the value. + unsafe { + match self.previous.take() { + Some(previous) => std::env::set_var(CAPTURE_TOOL_ARGUMENTS_ENV, previous), + None => std::env::remove_var(CAPTURE_TOOL_ARGUMENTS_ENV), + } + } + } +} + +const CAPTURE_TOOL_ARGUMENTS_ENV: &str = "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"; diff --git a/observability/grafana/datasources.yml b/observability/grafana/datasources.yml new file mode 100644 index 00000000..4372954a --- /dev/null +++ b/observability/grafana/datasources.yml @@ -0,0 +1,37 @@ +apiVersion: 1 + +# Starter data sources for a local Grafana stack. Replace the URLs and UIDs to +# match the deployment. The time shifts matter for OpenAPPA's sub-millisecond +# hook spans: without them, Grafana can ask Loki for an effectively empty time +# window when following a trace's "Related logs" link. +datasources: + - name: OpenAPPA Prometheus + type: prometheus + uid: prometheus + access: proxy + url: http://prometheus:9090 + editable: true + + - name: OpenAPPA Loki + type: loki + uid: loki + access: proxy + url: http://loki:3100 + editable: true + + - name: OpenAPPA Tempo + type: tempo + uid: tempo + access: proxy + url: http://tempo:3200 + editable: true + jsonData: + tracesToLogsV2: + datasourceUid: loki + filterByTraceID: true + filterBySpanID: false + spanStartTimeShift: -5m + spanEndTimeShift: 5m + tags: + - key: service.name + value: service_name diff --git a/observability/grafana/openappa-overview.json b/observability/grafana/openappa-overview.json new file mode 100644 index 00000000..0e835cd3 --- /dev/null +++ b/observability/grafana/openappa-overview.json @@ -0,0 +1,611 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "description": "OpenAPPA runtime hook decisions with trace-correlated logs, bounded metrics, and per-Agent-run drill-down.", + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "gridPos": { + "h": 4, + "w": 4, + "x": 0, + "y": 0 + }, + "id": 7, + "options": { + "code": { + "language": "plaintext", + "showLineNumbers": false, + "showMiniMap": false + }, + "content": "\"OpenAPPA\"", + "mode": "html" + }, + "transparent": true, + "type": "text" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "decimals": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 4, + "y": 0 + }, + "id": 1, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["sum"], + "fields": "", + "values": false + }, + "textMode": "auto", + "wideLayout": true + }, + "targets": [ + { + "limit": 20, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision =~ \"$decision\" } | count_over_time()", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Hook decisions", + "type": "stat" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "decimals": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 9, + "y": 0 + }, + "id": 2, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["sum"], + "fields": "", + "values": false + }, + "textMode": "auto", + "wideLayout": true + }, + "targets": [ + { + "limit": 20, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision =~ \"deny|block\" } | count_over_time()", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Policy stops", + "type": "stat" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "description": "Hook requests OpenAPPA could not parse or safely evaluate. Any non-zero value needs investigation.", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "decimals": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 14, + "y": 0 + }, + "id": 8, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["sum"], + "fields": "", + "values": false + }, + "textMode": "auto", + "wideLayout": true + }, + "targets": [ + { + "limit": 20, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision = \"refuse\" } | count_over_time()", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Runtime refusals", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${metrics}" + }, + "description": "Seconds since the newest reporting runtime started. No signal means telemetry has gone stale or no runtime is exporting.", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "decimals": 0, + "mappings": [ + { + "options": { + "-1": { + "color": "red", + "text": "No signal" + } + }, + "type": "value" + } + ], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": null + }, + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 19, + "y": 0 + }, + "id": 10, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto", + "wideLayout": true + }, + "targets": [ + { + "editorMode": "code", + "expr": "max(appa_runtime_uptime_seconds) or vector(-1)", + "instant": true, + "legendFormat": "Runtime uptime", + "range": false, + "refId": "A" + } + ], + "title": "Runtime uptime", + "type": "stat" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "decisions / interval", + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 12, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "showPoints": "never", + "spanNulls": true, + "stacking": { + "group": "A", + "mode": "none" + } + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 4 + }, + "id": 3, + "options": { + "legend": { + "calcs": ["sum"], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "limit": 20, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision =~ \"$decision\" } | count_over_time() by (span.appa.decision)", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Decision rate", + "type": "timeseries" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "seconds", + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 12, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "showPoints": "never", + "spanNulls": true, + "stacking": { + "group": "A", + "mode": "none" + } + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 4 + }, + "id": 4, + "options": { + "legend": { + "calcs": ["lastNotNull", "max"], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "limit": 20, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision =~ \"$decision\" } | quantile_over_time(span:duration, .95) by (span.appa.decision)", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Hook latency", + "type": "timeseries" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "fieldConfig": { + "defaults": {}, + "overrides": [ + { + "matcher": { + "id": "byRegexp", + "options": "^(Trace ID|traceID)$" + }, + "properties": [ + { + "id": "links", + "value": [ + { + "targetBlank": false, + "title": "Inspect this trace", + "url": "/d/openappa-runtime/openappa-runtime?var-traceId=${__value.raw}&var-agentRun=${agentRun}&var-decision=${decision}&var-trajectory=${trajectory}&var-metrics=${metrics}&var-traces=${traces}&var-logs=${logs}&${__url_time_range}" + } + ] + } + ] + } + ] + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 0, + "y": 12 + }, + "id": 5, + "options": { + "cellHeight": "sm", + "showHeader": true + }, + "targets": [ + { + "limit": 100, + "query": "{ resource.service.name = \"appa-runtime\" && resource.archestra.agent_run.id =~ \"$agentRun\" && span.appa.decision =~ \"$decision\" && span.appa.trajectory.id =~ \"$trajectory\" } | select(resource.archestra.agent_run.id, span.appa.trajectory.id, span.appa.hook.event, span.appa.decision, span.gen_ai.tool.name)", + "queryType": "traceql", + "refId": "A", + "tableType": "traces" + } + ], + "title": "Decision traces — select a trace for its spans and correlated logs", + "type": "table" + }, + { + "datasource": { + "type": "loki", + "uid": "${logs}" + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 0, + "y": 22 + }, + "id": 6, + "options": { + "dedupStrategy": "none", + "enableLogDetails": true, + "prettifyLogMessage": false, + "showCommonLabels": false, + "showLabels": false, + "showTime": true, + "sortOrder": "Descending", + "wrapLogMessage": true + }, + "targets": [ + { + "direction": "backward", + "editorMode": "code", + "expr": "{service_name=\"appa-runtime\"} | trace_id=~`$traceId` | appa_decision=~`$decision` | appa_trajectory_id=~`$trajectory`", + "queryType": "range", + "refId": "A" + } + ], + "title": "Correlated decision logs — trace $traceId", + "type": "logs" + }, + { + "datasource": { + "type": "tempo", + "uid": "${traces}" + }, + "description": "Trace waterfall for the selected decision. Click a Trace ID above to load it here and filter the correlated logs. Expand the appa.hook span to inspect gen_ai.tool.call.arguments when sensitive-content capture is enabled.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "gridPos": { + "h": 20, + "w": 12, + "x": 12, + "y": 12 + }, + "id": 9, + "options": {}, + "targets": [ + { + "query": "${traceId}", + "queryType": "traceql", + "refId": "A" + } + ], + "title": "Selected trace — $traceId", + "type": "traces" + } + ], + "refresh": "10s", + "schemaVersion": 41, + "tags": ["openappa", "security", "opentelemetry"], + "templating": { + "list": [ + { + "current": {}, + "label": "Metrics", + "name": "metrics", + "query": "prometheus", + "refresh": 1, + "type": "datasource" + }, + { + "current": {}, + "label": "Traces", + "name": "traces", + "query": "tempo", + "refresh": 1, + "type": "datasource" + }, + { + "current": {}, + "label": "Logs", + "name": "logs", + "query": "loki", + "refresh": 1, + "type": "datasource" + }, + { + "allValue": ".*", + "current": { + "text": "All", + "value": "$__all" + }, + "includeAll": true, + "label": "Decision", + "multi": true, + "name": "decision", + "options": [], + "query": "ack,allow,deny,block,refuse,pass_control,replace_output,replace_child_return", + "type": "custom" + }, + { + "current": { + "text": ".*", + "value": ".*" + }, + "description": "Platform Agent run UUID from /chat/runs/:id. This filters traces and their correlated logs without adding an unbounded metrics label.", + "label": "Agent run", + "name": "agentRun", + "query": ".*", + "type": "textbox" + }, + { + "current": { + "text": ".*", + "value": ".*" + }, + "label": "Trajectory regex", + "name": "trajectory", + "query": ".*", + "type": "textbox" + }, + { + "current": { + "text": ".*", + "value": ".*" + }, + "description": "Click a Trace ID in the decision table to load its waterfall and correlated logs.", + "label": "Trace ID", + "name": "traceId", + "query": ".*", + "type": "textbox" + } + ] + }, + "time": { + "from": "now-6h", + "to": "now" + }, + "timepicker": {}, + "timezone": "browser", + "title": "OpenAPPA Runtime", + "uid": "openappa-runtime", + "version": 1, + "weekStart": "" +} diff --git a/website/content/docs/observability.md b/website/content/docs/observability.md index c32273cb..63d6657f 100644 --- a/website/content/docs/observability.md +++ b/website/content/docs/observability.md @@ -4,3 +4,126 @@ category: Operations order: 9 description: What OpenAPPA emits in production and how to watch it. --- + +`appa runtime` can export traces, logs, and metrics over OTLP. Exporting is +off until an OpenTelemetry endpoint is configured, so the default runtime +still writes only to stderr. + +OpenAPPA's event log remains the authoritative security record. Operational +telemetry is a projection for finding slow decisions, refused hooks, and +unhealthy deployments. It also covers decisions that append no event-log +fact, such as a proposed call blocked before dispatch. + +## Enable OTLP + +Point the runtime at an OTLP/HTTP collector: + +```sh +OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4318 \ +OTEL_RESOURCE_ATTRIBUTES=deployment.environment.name=development \ +appa runtime -v --config appa.toml --db appa.db +``` + +The exporter follows the standard signal-specific OpenTelemetry variables, +including `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, +`OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`, +`OTEL_EXPORTER_OTLP_METRICS_ENDPOINT`, and +`OTEL_EXPORTER_OTLP_HEADERS`. Signal-specific settings override the common +endpoint. `OTEL_METRIC_EXPORT_INTERVAL` controls the metric export interval in +milliseconds. + +Tool arguments are sensitive content and are not exported by default. Set +`OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true` to add the exact JSON +arguments from call and result hooks to their spans as +`gen_ai.tool.call.arguments`. The runtime does not copy them to logs or +metrics. It exports arguments up to 32 KiB. For a larger value, the span carries +only `appa.tool.call.arguments.size_bytes` and +`appa.tool.call.arguments.sha256`. + +Use `-v` to include the runtime's decision path at debug level and `-vv` to +include the engine algebra at trace level. Without either flag, the decision +summary is still emitted at info level. Network export happens off the hook's +request path; an unavailable collector does not allow, block, or otherwise +change a tool call. + +On SIGTERM or Ctrl-C, the runtime stops accepting new requests and flushes all +three signals before exiting. + +## Signals + +Every parsed hook is represented by an `appa.hook` span and a structured +`hook decision` log inside that span. OTLP carries the trace and span IDs on +the log record, which lets a backend navigate from the decision trace to the +logs produced while that decision was evaluated. + +The span attributes are: + +| Attribute | Meaning | +|---|---| +| `appa.hook.event` | Bounded hook kind, such as `tool_call` or `tool_result` | +| `appa.decision` | Bounded outcome, such as `allow`, `deny`, `block`, or `refuse` | +| `appa.trajectory.id` | Root trajectory ID used to correlate separate hook requests | +| `appa.trajectory.child_id` | Child trajectory ID when the event belongs to one | +| `gen_ai.conversation.id` | The root trajectory under the OpenTelemetry GenAI convention | +| `gen_ai.tool.name` | Tool name on call and result hooks | +| `gen_ai.tool.call.arguments` | Exact tool arguments when sensitive-content capture is enabled and the value is at most 32 KiB | +| `appa.tool.call.arguments.size_bytes` | Argument size when an enabled capture exceeds 32 KiB | +| `appa.tool.call.arguments.sha256` | Argument digest when an enabled capture exceeds 32 KiB | +| `error.type` | Operational refusal family, when the span failed | + +The `hook decision` log repeats the event, decision, trajectory IDs, +conversation ID, and tool name. It does not repeat tool arguments or their +oversize metadata. Its trace and span IDs provide the correlation instead. + +The custom `appa.*` namespace is used because the OpenTelemetry registry has +no policy-decision or agent-trajectory convention. Standard `service.*`, +`deployment.*`, `error.type`, and `gen_ai.*` attributes are used where their +meaning matches. + +The runtime exports these low-cardinality metrics: + +| Metric | Type | Dimensions | +|---|---|---| +| `appa.runtime.hook.requests` | Counter | `appa.hook.event`, `appa.decision` | +| `appa.runtime.hook.duration` | Histogram, seconds | `appa.hook.event`, `appa.decision` | +| `appa.runtime.uptime` | Gauge, seconds | none | + +Trajectory IDs and tool names are intentionally absent from metric dimensions. +They remain available on spans and logs for drilldown without creating +unbounded time series. + +## Data boundary + +Telemetry never includes prompt text, tool results, policy feedback, remedy +text, or label values. Tool arguments are included on spans only when a +deployment explicitly enables sensitive-content capture. Arguments can contain +the data the policy exists to confine. Scope access to the observability backend +and set its retention period before enabling capture. A deployment may also +treat trajectory IDs and tool names as sensitive operational metadata. + +## Local Grafana + +Grafana's OpenTelemetry LGTM image provides an OTLP collector, Prometheus, +Loki, Tempo, and Grafana for local testing: + +```sh +docker run --rm --name appa-lgtm \ + -p 3002:3000 -p 4317:4317 -p 4318:4318 \ + grafana/otel-lgtm +``` + +Open [Grafana](http://localhost:3002) with `admin` / `admin`, then import +[`observability/grafana/openappa-overview.json`](https://github.com/archestra-ai/OpenAPPA/blob/main/observability/grafana/openappa-overview.json). +For a provisioned stack, the repository also includes a starter +[`datasources.yml`](https://github.com/archestra-ai/OpenAPPA/blob/main/observability/grafana/datasources.yml) +with Tempo-to-Loki correlation. Adjust its URLs and UIDs for the deployment. +Click a decision's trace ID to keep the investigation inside the dashboard: +the selected trace renders as a waterfall and filters the log panel to records +with the same trace ID. The dashboard's trajectory field widens that view +across all hook traces belonging to one trajectory. + +The top row separates expected policy stops (`deny` and `block`) from runtime +refusals, which indicate a hook OpenAPPA could not safely evaluate. Runtime +uptime supplies a heartbeat for deployment-specific missing-telemetry alerts. +Alert notification routes and thresholds belong to the deployment's Grafana +configuration; the starter dashboard does not choose recipients.