Apps/ImageApi
1
0 Fork 0
Code Issues 7 Pull Requests Releases Wiki Activity

Tracing: LLM clients + streaming spans on tracing, drop cx threading (#140)

Browse Source 
LlmClient methods no longer take an OpenTelemetry Context. llm.request is
now a tracing span (otel.kind=client, llm.provider/operation/model) that
parents on the caller's ambient span; traceparent injection reads the
span's OTel context via OpenTelemetrySpanExt.

The src/ai feature spans (handlers, insight generation, chat, persona
chat, daily summaries, TTS) move to #[instrument] / info_span!, and every
LLM-bearing tokio::spawn carries its parent span with .instrument(): the
streaming chat turn (chat_turn_stream_inner), async chat / persona turns,
background insight jobs and TTS jobs. Sub-spans in the agentic loops
(ai.chat.loop, ai.agentic.loop) instrument each LLM / tool await.

DAO calls keep their explicit Context (derived from the tracing span)
until the DAO migration (#141).

tests/otel_llm_spans.rs pins nesting + traceparent for the blocking and
streaming paths across a spawn on a multi-thread runtime.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-09-29 17:02:40 -04:00
1 parent 7a59afc62b
commit 1c508cdabd
20 files changed
+973 -1030
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/ai/handlers.rs
+205 -212
View File
@@ -1,8 +1,9 @@
use actix_web::{HttpRequest, HttpResponse, Responder, delete, get, post, web};
use actix_web::{HttpResponse, Responder, delete, get, post, web};
use futures::StreamExt;
use opentelemetry::KeyValue;
use opentelemetry::trace::{Span, Status, Tracer};
use opentelemetry::trace::Status;
use serde::{Deserialize, Serialize};
use tracing::Instrument;
use tracing_opentelemetry::OpenTelemetrySpanExt;
use crate::ai::insight_chat::{ChatStreamEvent, ChatTurnRequest};
use crate::ai::ollama::ChatMessage;
@@ -11,7 +12,6 @@ use crate::data::Claims;
use crate::database::models::{InsightGenerationType, InsightJobStatus, PhotoInsight};
use crate::database::{ExifDao, InsightDao};
use crate::libraries;
use crate::otel::{extract_context_from_request, global_tracer};
use crate::state::AppState;
use crate::utils::normalize_path;
@@ -416,15 +416,13 @@ fn apply_owner_defaults(
/// POST /insights/generate - Generate insight for a specific photo (async)
#[post("/insights/generate")]
#[tracing::instrument(name = "http.insights.generate", skip_all)]
pub async fn generate_insight_handler(
http_request: HttpRequest,
_claims: Claims,
request: web::Json<GeneratePhotoInsightRequest>,
app_state: web::Data<AppState>,
) -> impl Responder {
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("http.insights.generate", &parent_context);
let span = tracing::Span::current();
let mut request = request.0;
// Stored owner defaults fill what the client omitted (request → pref →
@@ -440,9 +438,9 @@ pub async fn generate_insight_handler(
let library = app_state.primary_library();
let gen_type = InsightGenerationType::Standard;
span.set_attribute(KeyValue::new("file_path", normalized_path.clone()));
span.set_attribute("file_path", normalized_path.clone());
if let Some(ref model) = request.model {
span.set_attribute(KeyValue::new("model", model.clone()));
span.set_attribute("model", model.clone());
}
log::info!(
@@ -506,96 +504,102 @@ pub async fn generate_insight_handler(
let job_handles = app_state.insight_job_handles.clone();
let path = normalized_path.clone();
let handle = tokio::spawn(async move {
let timeout_secs: u64 = std::env::var("INSIGHT_GENERATION_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(120);
let handle = tokio::spawn(
async move {
let timeout_secs: u64 = std::env::var("INSIGHT_GENERATION_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(120);
let path_for_task = path.clone();
let generator_for_task = generator.clone();
let result = tokio::task::spawn(async move {
// Cross-model barrier: if a TTS synthesis holds the GPU, wait it
// out BEFORE the generation wall-clock starts. The per-request
// lease keeps reqwest budgets honest, but this job-level timeout
// would otherwise burn while the first chat call queues behind a
// multi-minute synthesis. Dropped immediately — holding it across
// the generation would deadlock the chat calls' own leases.
drop(crate::ai::gpu::llm_lease().await);
tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
generator_for_task.generate_insight_for_photo_with_config(
&path_for_task,
request.model.clone(),
request.system_prompt.clone(),
request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
),
)
.await
})
.await;
let path_for_task = path.clone();
let generator_for_task = generator.clone();
let result = tokio::task::spawn(async move {
// Cross-model barrier: if a TTS synthesis holds the GPU, wait it
// out BEFORE the generation wall-clock starts. The per-request
// lease keeps reqwest budgets honest, but this job-level timeout
// would otherwise burn while the first chat call queues behind a
// multi-minute synthesis. Dropped immediately — holding it across
// the generation would deadlock the chat calls' own leases.
drop(crate::ai::gpu::llm_lease().await);
tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
generator_for_task.generate_insight_for_photo_with_config(
&path_for_task,
request.model.clone(),
request.system_prompt.clone(),
request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
),
)
.await
})
.await;
let ctx = opentelemetry::Context::new();
let mut dao = job_dao.lock().expect("Unable to lock InsightJobDao");
let ctx = opentelemetry::Context::new();
let mut dao = job_dao.lock().expect("Unable to lock InsightJobDao");
match result {
Ok(Ok(Ok(()))) => {
let mut insight_dao = generator
.insight_dao()
.lock()
.expect("Unable to lock InsightDao");
let insight_id = insight_dao
.get_insight(&ctx, &path)
.ok()
.flatten()
.map(|i| i.id);
if let Some(id) = insight_id {
if let Err(e) = dao.complete_job(&ctx, job_id, id) {
log::error!("Failed to mark job {} as completed: {:?}", job_id, e);
match result {
Ok(Ok(Ok(()))) => {
let mut insight_dao = generator
.insight_dao()
.lock()
.expect("Unable to lock InsightDao");
let insight_id = insight_dao
.get_insight(&ctx, &path)
.ok()
.flatten()
.map(|i| i.id);
if let Some(id) = insight_id {
if let Err(e) = dao.complete_job(&ctx, job_id, id) {
log::error!("Failed to mark job {} as completed: {:?}", job_id, e);
}
} else if let Err(e) =
dao.fail_job(&ctx, job_id, "generation returned no insight")
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, e);
}
}
Ok(Ok(Err(e))) => {
log::error!("Insight generation failed for {}: {:?}", path, e);
// `{:#}` = one-line context chain; the job's error_message is
// returned to the client verbatim, so no Debug/backtrace here.
if let Err(err) = dao.fail_job(&ctx, job_id, &format!("{:#}", e)) {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
Ok(Err(_)) => {
log::error!(
"Insight generation timed out for {} after {}s",
path,
timeout_secs
);
if let Err(err) =
dao.fail_job(&ctx, job_id, &format!("timeout after {}s", timeout_secs))
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
Err(_) => {
log::error!("Insight generation task panicked for {}", path);
if let Err(err) = dao.fail_job(&ctx, job_id, "generation task panicked") {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
} else if let Err(e) = dao.fail_job(&ctx, job_id, "generation returned no insight")
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, e);
}
}
Ok(Ok(Err(e))) => {
log::error!("Insight generation failed for {}: {:?}", path, e);
// `{:#}` = one-line context chain; the job's error_message is
// returned to the client verbatim, so no Debug/backtrace here.
if let Err(err) = dao.fail_job(&ctx, job_id, &format!("{:#}", e)) {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
Ok(Err(_)) => {
log::error!(
"Insight generation timed out for {} after {}s",
path,
timeout_secs
);
if let Err(err) =
dao.fail_job(&ctx, job_id, &format!("timeout after {}s", timeout_secs))
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
Err(_) => {
log::error!("Insight generation task panicked for {}", path);
if let Err(err) = dao.fail_job(&ctx, job_id, "generation task panicked") {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
}
// Remove handle from map on completion
let mut handles = job_handles
.lock()
.expect("Unable to lock InsightJobHandles");
handles.remove(&job_id);
});
// Remove handle from map on completion
let mut handles = job_handles
.lock()
.expect("Unable to lock InsightJobHandles");
handles.remove(&job_id);
}
// Carry the handler span into the job so the generation (and its
// LLM / DB spans) stays in the request's trace.
.instrument(tracing::Span::current()),
);
// Store abort handle
{
@@ -606,7 +610,7 @@ pub async fn generate_insight_handler(
handles.insert(job_id, handle.abort_handle());
}
span.set_attribute(KeyValue::new("job_id", job_id as i64));
span.set_attribute("job_id", job_id as i64);
span.set_status(Status::Ok);
HttpResponse::Accepted().json(JobIdResponse { job_id })
}
@@ -744,15 +748,13 @@ pub async fn get_insight_history_handler(
/// POST /insights/generate/agentic - Generate insight using agentic tool-calling loop (async)
#[post("/insights/generate/agentic")]
#[tracing::instrument(name = "http.insights.generate_agentic", skip_all)]
pub async fn generate_agentic_insight_handler(
http_request: HttpRequest,
claims: Claims,
request: web::Json<GeneratePhotoInsightRequest>,
app_state: web::Data<AppState>,
) -> impl Responder {
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("http.insights.generate_agentic", &parent_context);
let span = tracing::Span::current();
let mut request = request.0;
// Stored owner defaults fill what the client omitted (request → pref →
@@ -768,12 +770,12 @@ pub async fn generate_agentic_insight_handler(
let library = app_state.primary_library();
let gen_type = InsightGenerationType::Agentic;
span.set_attribute(KeyValue::new("file_path", normalized_path.clone()));
span.set_attribute("file_path", normalized_path.clone());
if let Some(ref model) = request.model {
span.set_attribute(KeyValue::new("model", model.clone()));
span.set_attribute("model", model.clone());
}
if let Some(ref backend) = request.backend {
span.set_attribute(KeyValue::new("backend", backend.clone()));
span.set_attribute("backend", backend.clone());
}
log::info!(
@@ -901,91 +903,97 @@ pub async fn generate_agentic_insight_handler(
let job_handles = app_state.insight_job_handles.clone();
let path = normalized_path.clone();
let handle = tokio::spawn(async move {
let timeout_secs: u64 = std::env::var("INSIGHT_GENERATION_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(180);
let handle = tokio::spawn(
async move {
let timeout_secs: u64 = std::env::var("INSIGHT_GENERATION_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(180);
let path_for_task = path.clone();
let generator_for_task = generator.clone();
let result = tokio::task::spawn(async move {
// Cross-model barrier — see generate_insight_handler: wait out any
// running TTS synthesis before the generation wall-clock starts.
drop(crate::ai::gpu::llm_lease().await);
tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
generator_for_task.generate_agentic_insight_for_photo(
&path_for_task,
request.model.clone(),
system_prompt,
request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
request.enable_thinking,
max_iterations,
request.backend.clone(),
fewshot_examples,
fewshot_ids,
user_id,
persona_id,
),
)
.await
})
.await;
let path_for_task = path.clone();
let generator_for_task = generator.clone();
let result = tokio::task::spawn(async move {
// Cross-model barrier — see generate_insight_handler: wait out any
// running TTS synthesis before the generation wall-clock starts.
drop(crate::ai::gpu::llm_lease().await);
tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
generator_for_task.generate_agentic_insight_for_photo(
&path_for_task,
request.model.clone(),
system_prompt,
request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
request.enable_thinking,
max_iterations,
request.backend.clone(),
fewshot_examples,
fewshot_ids,
user_id,
persona_id,
),
)
.await
})
.await;
let ctx = opentelemetry::Context::new();
let mut dao = job_dao.lock().expect("Unable to lock InsightJobDao");
let ctx = opentelemetry::Context::new();
let mut dao = job_dao.lock().expect("Unable to lock InsightJobDao");
match result {
Ok(Ok(Ok((Some(insight_id), _, _)))) => {
if let Err(e) = dao.complete_job(&ctx, job_id, insight_id) {
log::error!("Failed to mark job {} as completed: {:?}", job_id, e);
match result {
Ok(Ok(Ok((Some(insight_id), _, _)))) => {
if let Err(e) = dao.complete_job(&ctx, job_id, insight_id) {
log::error!("Failed to mark job {} as completed: {:?}", job_id, e);
}
}
}
Ok(Ok(Ok((None, _, _)))) => {
if let Err(e) = dao.fail_job(&ctx, job_id, "agentic generation returned no insight")
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, e);
Ok(Ok(Ok((None, _, _)))) => {
if let Err(e) =
dao.fail_job(&ctx, job_id, "agentic generation returned no insight")
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, e);
}
}
}
Ok(Ok(Err(e))) => {
log::error!("Agentic insight generation failed for {}: {:?}", path, e);
// `{:#}` = one-line context chain; the job's error_message is
// returned to the client verbatim, so no Debug/backtrace here.
if let Err(err) = dao.fail_job(&ctx, job_id, &format!("{:#}", e)) {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
Ok(Ok(Err(e))) => {
log::error!("Agentic insight generation failed for {}: {:?}", path, e);
// `{:#}` = one-line context chain; the job's error_message is
// returned to the client verbatim, so no Debug/backtrace here.
if let Err(err) = dao.fail_job(&ctx, job_id, &format!("{:#}", e)) {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
}
Ok(Err(_)) => {
log::error!(
"Agentic insight generation timed out for {} after {}s",
path,
timeout_secs
);
if let Err(err) =
dao.fail_job(&ctx, job_id, &format!("timeout after {}s", timeout_secs))
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
Ok(Err(_)) => {
log::error!(
"Agentic insight generation timed out for {} after {}s",
path,
timeout_secs
);
if let Err(err) =
dao.fail_job(&ctx, job_id, &format!("timeout after {}s", timeout_secs))
{
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
}
Err(_) => {
log::error!("Agentic insight generation task panicked for {}", path);
if let Err(err) = dao.fail_job(&ctx, job_id, "generation task panicked") {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
Err(_) => {
log::error!("Agentic insight generation task panicked for {}", path);
if let Err(err) = dao.fail_job(&ctx, job_id, "generation task panicked") {
log::error!("Failed to mark job {} as failed: {:?}", job_id, err);
}
}
}
// Remove handle from map on completion
let mut handles = job_handles
.lock()
.expect("Unable to lock InsightJobHandles");
handles.remove(&job_id);
}
// Remove handle from map on completion
let mut handles = job_handles
.lock()
.expect("Unable to lock InsightJobHandles");
handles.remove(&job_id);
});
// Carry the handler span into the job so the generation (and its
// LLM / DB spans) stays in the request's trace.
.instrument(tracing::Span::current()),
);
// Store abort handle
{
@@ -996,7 +1004,7 @@ pub async fn generate_agentic_insight_handler(
handles.insert(job_id, handle.abort_handle());
}
span.set_attribute(KeyValue::new("job_id", job_id as i64));
span.set_attribute("job_id", job_id as i64);
span.set_status(Status::Ok);
HttpResponse::Accepted().json(JobIdResponse { job_id })
}
@@ -1270,16 +1278,14 @@ pub struct ChatTurnHttpResponse {
/// POST /insights/chat — submit a follow-up turn against an existing insight.
#[post("/insights/chat")]
#[tracing::instrument(name = "http.insights.chat", skip_all)]
pub async fn chat_turn_handler(
http_request: HttpRequest,
claims: Claims,
request: web::Json<ChatTurnHttpRequest>,
app_state: web::Data<AppState>,
) -> impl Responder {
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("http.insights.chat", &parent_context);
span.set_attribute(KeyValue::new("file_path", request.file_path.clone()));
let span = tracing::Span::current();
span.set_attribute("file_path", request.file_path.clone());
let mut request = request.0;
// Stored owner defaults fill what the client omitted (request → pref →
@@ -1667,12 +1673,10 @@ pub struct ChatSwitchBranchRequest {
/// Returns `text/event-stream` with one event per chat stream event.
#[post("/insights/chat/stream")]
pub async fn chat_stream_handler(
http_request: HttpRequest,
claims: Claims,
request: web::Json<ChatTurnHttpRequest>,
app_state: web::Data<AppState>,
) -> HttpResponse {
let parent_context = extract_context_from_request(&http_request);
let mut request = request.0;
// Stored owner defaults fill what the client omitted (request → pref →
// hard default; the hard default still applies downstream).
@@ -1718,7 +1722,7 @@ pub async fn chat_stream_handler(
};
let service = app_state.insight_chat.clone();
let events = service.chat_turn_stream(chat_req, parent_context);
let events = service.chat_turn_stream(chat_req);
// Map ChatStreamEvent → SSE frame bytes.
let sse_stream = futures::stream::StreamExt::map(events, |ev| {
@@ -1819,16 +1823,14 @@ fn sse_event_payload(ev: &ChatStreamEvent) -> (&'static str, serde_json::Value)
/// POST /insights/chat/turn — async turn dispatch. Returns turn_id immediately,
/// client then polls GET /insights/chat/turn/{turn_id} for SSE replay.
#[post("/insights/chat/turn")]
#[tracing::instrument(name = "http.insights.chat_turn_async", skip_all)]
pub async fn turn_async_handler(
http_request: HttpRequest,
claims: Claims,
request: web::Json<ChatTurnHttpRequest>,
app_state: web::Data<AppState>,
) -> impl Responder {
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("http.insights.chat_turn_async", &parent_context);
span.set_attribute(KeyValue::new("file_path", request.file_path.clone()));
let span = tracing::Span::current();
span.set_attribute("file_path", request.file_path.clone());
let mut request = request.0;
// Stored owner defaults fill what the client omitted (request → pref →
@@ -1877,7 +1879,7 @@ pub async fn turn_async_handler(
let registry = app_state.turn_registry.clone();
let turn_id = service.chat_turn_async(registry, chat_req).await;
span.set_attribute(KeyValue::new("turn_id", turn_id.clone()));
span.set_attribute("turn_id", turn_id.clone());
span.set_status(Status::Ok);
HttpResponse::Accepted().json(serde_json::json!({
"turn_id": turn_id,
@@ -1896,13 +1898,13 @@ pub struct ReplayQuery {
/// GET /insights/chat/turn/{turn_id} — SSE replay stream.
#[get("/insights/chat/turn/{turn_id}")]
#[tracing::instrument(name = "ai.chat.turn.replay", skip_all)]
pub async fn turn_replay_handler(
http_request: HttpRequest,
path: web::Path<String>,
query: web::Query<ReplayQuery>,
app_state: web::Data<AppState>,
) -> HttpResponse {
turn_replay_impl(http_request, path, query, app_state).await
turn_replay_impl(path, query, app_state).await
}
/// Core of the SSE replay, kept attribute-free so the persona-chat routes
@@ -1910,7 +1912,6 @@ pub async fn turn_replay_handler(
/// on `turn_id` only, and `render_turn_info_frame` already scopes the
/// identity (persona turns carry `persona_id` and no `file_path`).
pub(crate) async fn turn_replay_impl(
http_request: HttpRequest,
path: web::Path<String>,
query: web::Query<ReplayQuery>,
app_state: web::Data<AppState>,
@@ -1920,11 +1921,9 @@ pub(crate) async fn turn_replay_impl(
let turn_id = path.into_inner();
let skip_before = query.skip_before.unwrap_or(0);
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("ai.chat.turn.replay", &parent_context);
span.set_attribute(KeyValue::new("turn_id", turn_id.clone()));
span.set_attribute(KeyValue::new("skip_before", skip_before as i64));
let span = tracing::Span::current();
span.set_attribute("turn_id", turn_id.clone());
span.set_attribute("skip_before", skip_before as i64);
let registry = app_state.turn_registry.clone();
let entry = match registry.get(&turn_id).await {
@@ -1938,11 +1937,8 @@ pub(crate) async fn turn_replay_impl(
};
let info = entry.info().await;
span.set_attribute(KeyValue::new("status", info.status.as_str()));
span.set_attribute(KeyValue::new(
"event_count",
info.total_events_pushed as i64,
));
span.set_attribute("status", info.status.as_str());
span.set_attribute("event_count", info.total_events_pushed as i64);
let turn_info_frame = render_turn_info_frame(&info);
// Initial buffered batch: events produced before this connection attached.
@@ -2074,27 +2070,24 @@ fn render_turn_info_frame(info: &crate::ai::turn_registry::TurnInfo) -> String {
/// DELETE /insights/chat/turn/{turn_id} — cancel a running turn.
#[delete("/insights/chat/turn/{turn_id}")]
#[tracing::instrument(name = "ai.chat.turn.cancel", skip_all)]
pub async fn cancel_turn_handler(
http_request: HttpRequest,
path: web::Path<String>,
app_state: web::Data<AppState>,
) -> impl Responder {
cancel_turn_impl(http_request, path, app_state).await
cancel_turn_impl(path, app_state).await
}
/// Core of the turn-cancel, attribute-free so the persona-chat DELETE route
/// can reuse it under `/persona_chat/turn/{turn_id}`.
pub(crate) async fn cancel_turn_impl(
http_request: HttpRequest,
path: web::Path<String>,
app_state: web::Data<AppState>,
) -> impl Responder {
let turn_id = path.into_inner();
let parent_context = extract_context_from_request(&http_request);
let tracer = global_tracer();
let mut span = tracer.start_with_context("ai.chat.turn.cancel", &parent_context);
span.set_attribute(KeyValue::new("turn_id", turn_id.clone()));
let span = tracing::Span::current();
span.set_attribute("turn_id", turn_id.clone());
let registry = app_state.turn_registry.clone();
let entry = match registry.get(&turn_id).await {
@@ -2111,7 +2104,7 @@ pub(crate) async fn cancel_turn_impl(
// also checks `is_running()` at each iteration boundary as a graceful
// backstop in case the abort lands between await points.
let aborted = entry.abort();
span.set_attribute(KeyValue::new("aborted", aborted));
span.set_attribute("aborted", aborted);
// Push the terminal event BEFORE flipping status: a replay reader treats a
// terminal status with no buffered tail as "closed", so the Done must be