refactor: split row compute engine modules

Cargo.toml

@@ -130,6 +130,7 @@ sqlx = { version = "0.8.6", default-features = false, features = [
     "postgres",
     "runtime-tokio-rustls",
 ] }
+sqlx-tracing = { version = "0.2.1", default-features = false, features = ["postgres"] }
 redis = { version = "0.27", default-features = false, features = [
     "tokio-comp",
     "connection-manager",
@@ -165,6 +166,16 @@ tempfile = "3.0"
 chrono = { version = "0.4.44", default-features = false, features = ["clock", "serde"] }
 chrono-english = "0.1.8"
 rust_decimal = { version = "1.37", default-features = false, features = ["std", "serde", "serde-with-str"] }
+polars = { version = "0.46", default-features = false, features = [
+    "lazy",
+    "dtype-struct",
+    "dtype-decimal",
+    "strings",
+    "temporal",
+    "is_in",
+    "abs",
+    "regex",
+] }
 toon = "0.1"
 
 # LLM / REPL

apis/architect-exchange/domain.yaml

@@ -277,7 +277,7 @@ entities:
     id_from:
     - account_id
     - symbol
-    description: Signed derivative exposure for one instrument in an account.
+    description: Signed derivative exposure for one instrument in an account. Ad-hoc ranking and money sums of fetched positions use program postfix, not extra catalog entities.
     discovery:
       names:
       - position
@@ -940,7 +940,7 @@ entities:
 
   Fill:
     id_field: trade_id
-    description: Private execution against an account, including fees and side.
+    description: Private execution against an account, including fees and side. Ad-hoc rollups (fees by symbol, notional) use program postfix on fetched rows, not extra catalog entities.
     discovery:
       names:
       - fill

crates/plasm-agent-core/Cargo.toml

@@ -67,13 +67,15 @@ subtle = { workspace = true }
 # even when we only use `InMemoryStorage` at runtime (see auth_framework::storage::memory).
 auth-framework = { workspace = true }
 sqlx = { workspace = true, features = ["migrate", "chrono", "uuid"] }
+sqlx-tracing = { workspace = true }
 opentelemetry = { workspace = true }
 minijinja = { version = "2.19.0", default-features = false, features = ["builtins", "serde"] }
 futures = { workspace = true }
 redis = { workspace = true }
 dashmap = { workspace = true }
 
 [dev-dependencies]
+plasm-otel = { path = "../plasm-otel", features = ["testing"] }
 dhat = "0.3"
 tempfile = { workspace = true }
 insta = { workspace = true, features = ["json"] }

crates/plasm-agent-core/src/blocking_compute.rs

@@ -52,7 +52,7 @@ impl BlockingComputePool {
             .acquire()
             .await
             .map_err(|_| ComputePoolError::Closed)?;
-        let span = tracing::debug_span!("plasm.blocking_compute", label);
+        let span = crate::spans::blocking_compute(label);
         let out = tokio::task::spawn_blocking(move || {
             let _guard = span.enter();
             f()

crates/plasm-agent-core/src/flow_policy_repository.rs

@@ -1,9 +1,10 @@
 //! sqlx-backed persistence for project-scoped plan flow policies.
 
+use crate::traced_pg::PgPool;
 use chrono::{DateTime, Utc};
 use serde_json::Value;
 use sqlx::postgres::PgPoolOptions;
-use sqlx::{PgPool, Row};
+use sqlx::Row;
 use thiserror::Error;
 
 use crate::plan_flow_policy::{FlowPolicy, FlowPolicySnapshot, PolicyRevision};
@@ -65,6 +66,7 @@ impl FlowPolicyRepository {
             .connect(database_url)
             .await?;
         sqlx::migrate!("./migrations").run(&pool).await?;
+        let pool = crate::traced_pg::wrap(pool);
         Ok(Self { pool })
     }
 

crates/plasm-agent-core/src/graph_rehydrate/rehydrator.rs

@@ -13,7 +13,9 @@ use crate::execute_session::ExecuteSession;
 use crate::server_state::PlasmHostState;
 
 use super::ctx::GraphSurfaceWalkCtx;
-use super::walk::{collect_entities, collect_row_json, snapshot_hot_entities, stream_rows};
+#[cfg(test)]
+use super::walk::stream_rows;
+use super::walk::{collect_entities, collect_row_json, snapshot_hot_entities};
 
 /// Hot-cache snapshot + target count for spill rehydrate after the graph lock is released.
 pub(crate) struct GraphSpillSyncPlan {
@@ -233,6 +235,7 @@ impl<'a> GraphSurfaceRehydrator<'a> {
         self.rehydrate_rows(hot, entity_type, logical_count).await
     }
 
+    #[cfg(test)]
     pub(crate) async fn stream_entity_rows<F>(
         &self,
         hot_snapshot: Arc<[CachedEntity]>,

crates/plasm-agent-core/src/graph_rehydrate/walk.rs

@@ -125,6 +125,7 @@ where
     })
 }
 
+#[cfg(test)]
 pub(crate) async fn stream_rows<F>(
     ctx: &GraphSurfaceWalkCtx<'_>,
     hot_snapshot: Arc<[CachedEntity]>,
@@ -180,7 +181,7 @@ pub(crate) async fn collect_entities(
     out.truncate(logical_count);
     crate::graph_cache_metrics::record_graph_rehydrate(
         "full",
-        out.len(),
+        stats.rows_yielded,
         stats.pages_read,
         started.elapsed(),
     );

crates/plasm-agent-core/src/http.rs

@@ -224,7 +224,7 @@ pub fn discovery_execute_router(state: PlasmHostState) -> Router {
 
     let traced = pre_internal
         .merge(oss_traced_routes())
-        .layer(TraceLayer::new_for_http().make_span_with(tower_http_trace_parent_span));
+        .route_layer(TraceLayer::new_for_http().make_span_with(tower_http_trace_parent_span));
 
     Router::new()
         .merge(health_public)
@@ -285,7 +285,7 @@ pub async fn serve_discovery_execute_on_listener_opts(
 ) -> Result<(), Box<dyn std::error::Error>> {
     let addr = listener.local_addr()?;
     let port = addr.port();
-    tracing::info!("plasm HTTP listening on http://{addr}");
+    tracing::info!(%addr, "plasm HTTP listening");
     if opts.emit_stderr_route_help {
         eprint_http_command_help(port);
     }
@@ -317,7 +317,7 @@ pub async fn serve_discovery_execute_and_mcp_unified(
 ) -> Result<(), Box<dyn std::error::Error>> {
     let addr = listener.local_addr()?;
     let port = addr.port();
-    tracing::info!("plasm HTTP+MCP unified listening on http://{addr}");
+    tracing::info!(%addr, "plasm HTTP+MCP unified listening");
     if opts.emit_stderr_route_help {
         eprint_http_command_help(port);
     }

crates/plasm-agent-core/src/http_discovery.rs

@@ -25,6 +25,7 @@ use crate::server_state::{PlasmHostState, ToolModelHostError};
 use crate::tool_model::ToolModelQuery;
 use crate::tool_model_service::ToolModelServiceError;
 use crate::typed_discovery_host::run_typed_catalog_discovery;
+use tracing::Instrument;
 
 #[derive(Debug, Deserialize)]
 pub struct IncludeCgsQuery {
@@ -229,6 +230,15 @@ fn log_discovery_response(out: &DiscoveryResult) {
 async fn post_discover_typed(
     Extension(st): Extension<PlasmHostState>,
     Json(query): Json<DiscoveryQuery>,
+) -> Response {
+    post_discover_typed_inner(st, query)
+        .instrument(crate::spans::discover_query())
+        .await
+}
+
+pub(crate) async fn post_discover_typed_inner(
+    st: PlasmHostState,
+    query: DiscoveryQuery,
 ) -> Response {
     tracing::debug!(
         utterance_len = query.utterance.len(),
@@ -303,6 +313,12 @@ async fn post_discover(
     Extension(st): Extension<PlasmHostState>,
     Json(query): Json<CapabilityQuery>,
 ) -> Response {
+    post_discover_inner(st, query)
+        .instrument(crate::spans::discover_query())
+        .await
+}
+
+pub(crate) async fn post_discover_inner(st: PlasmHostState, query: CapabilityQuery) -> Response {
     tracing::debug!(
         tokens = query.tokens.len(),
         phrases = query.phrases.len(),

crates/plasm-agent-core/src/http_execute/routes/handlers/mod.rs

@@ -12,5 +12,7 @@ pub(crate) use artifacts::{
 };
 pub(crate) use create::post_create_execute_session;
 pub(crate) use run_post::post_run_execute_session;
+#[cfg(test)]
+pub(crate) use run_post::post_run_execute_session_inner;
 pub(crate) use session_get::handle_execute_session_get;
 pub(crate) use stream::get_operation_progress_stream;

crates/plasm-agent-core/src/http_execute/routes/handlers/run_post.rs

@@ -1,5 +1,7 @@
 //! POST run execute session.
 
+use tracing::Instrument;
+
 use super::super::super::*;
 
 use super::plan_run_response::respond_plan_run_live_result;
@@ -14,6 +16,20 @@ pub(crate) async fn post_run_execute_session(
     Query(run_q): Query<ExecuteRunQuery>,
     headers: HeaderMap,
     body: Bytes,
+) -> Response {
+    post_run_execute_session_inner(st, principal, prompt_hash, session_id, run_q, headers, body)
+        .instrument(crate::spans::execute_run_post())
+        .await
+}
+
+pub(crate) async fn post_run_execute_session_inner(
+    st: PlasmHostState,
+    principal: Option<crate::incoming_auth::TenantPrincipal>,
+    prompt_hash: PromptHashHex,
+    session_id: ExecuteSessionId,
+    run_q: ExecuteRunQuery,
+    headers: HeaderMap,
+    body: Bytes,
 ) -> Response {
     let Some(sess) = st
         .get_execute_session(prompt_hash.as_str(), session_id.as_str())
@@ -49,15 +65,15 @@ pub(crate) async fn post_run_execute_session(
         Ok(k) => k,
         Err(AcceptNegotiationError::NoSupportedMediaType) => {
             return problem_response(
-                Problem::custom(
-                    ProblemStatus::NOT_ACCEPTABLE,
-                    Uri::from_static(problem_types::EXECUTE_UNSUPPORTED_ACCEPT),
-                )
-                .with_title("Not Acceptable")
-                .with_detail(
-                    "supported Accept values include application/json, application/x-ndjson, text/plain, text/toon (default when Accept is omitted: text/toon)",
-                ),
-            );
+            Problem::custom(
+                ProblemStatus::NOT_ACCEPTABLE,
+                Uri::from_static(problem_types::EXECUTE_UNSUPPORTED_ACCEPT),
+            )
+            .with_title("Not Acceptable")
+            .with_detail(
+                "supported Accept values include application/json, application/x-ndjson, text/plain, text/toon (default when Accept is omitted: text/toon)",
+            ),
+        );
         }
     };
 
@@ -320,15 +336,15 @@ pub(crate) async fn post_run_execute_session(
         crate::PlanGateDecision::Proceed(_) => {}
         crate::PlanGateDecision::NeedsReview => {
             return problem_response(
-                Problem::custom(
-                    ProblemStatus::BAD_REQUEST,
-                    Uri::from_static(problem_types::EXECUTE_INVALID_EXPRESSION),
-                )
-                .with_title("Bad Request")
-                .with_detail(
-                    "plan_requires_review: call plan dry-run first, then pass plan_commit_ref or force=true",
-                ),
-            );
+            Problem::custom(
+                ProblemStatus::BAD_REQUEST,
+                Uri::from_static(problem_types::EXECUTE_INVALID_EXPRESSION),
+            )
+            .with_title("Bad Request")
+            .with_detail(
+                "plan_requires_review: call plan dry-run first, then pass plan_commit_ref or force=true",
+            ),
+        );
         }
         crate::PlanGateDecision::Denied(denial) => {
             return problem_response(

crates/plasm-agent-core/src/http_execute/routes/tests.rs

@@ -866,3 +866,65 @@ fn negotiate_accept_variants() {
     );
     assert!(negotiate_accept(Some("application/soap+xml")).is_err());
 }
+
+#[test]
+fn http_request_parents_execute_run_post_on_handler() {
+    use crate::execute_path_ids::{ExecuteSessionId, PromptHashHex};
+    use crate::http_execute::response::ExecuteRunQuery;
+    use axum::body::Bytes;
+    use axum::http::HeaderMap;
+    use plasm_otel::span_capture::{find_span, is_child_of, with_captured_spans};
+    use plasm_otel::tower_http_trace_parent_span;
+    use tracing::Instrument;
+
+    let req = Request::builder()
+        .method("POST")
+        .uri("/execute/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+        .body(())
+        .unwrap();
+
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let ((), spans) = with_captured_spans(|| {
+        let http = tower_http_trace_parent_span(&req);
+        // Create `execute_run_post` only after `http.request` is entered — tracing records
+        // parent at span construction time (mirrors TraceLayer then handler `.instrument`).
+        rt.block_on(
+            async {
+                async {
+                    let st = test_state_with_registry();
+                    let prompt_hash: PromptHashHex =
+                        "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+                            .parse()
+                            .expect("prompt hash");
+                    let session_id: ExecuteSessionId = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+                        .parse()
+                        .expect("session id");
+                    let _ = super::handlers::post_run_execute_session_inner(
+                        st,
+                        None,
+                        prompt_hash,
+                        session_id,
+                        ExecuteRunQuery::default(),
+                        HeaderMap::new(),
+                        Bytes::new(),
+                    )
+                    .await;
+                }
+                .instrument(crate::spans::execute_run_post())
+                .await;
+            }
+            .instrument(http),
+        );
+    });
+
+    let parent = find_span(&spans, "plasm_agent.http.request").expect("http.request");
+    let child = find_span(&spans, "plasm_agent.execute.run_post").expect("execute.run_post");
+    assert!(
+        is_child_of(child, parent),
+        "execute.run_post must be child of http.request; spans={:?}",
+        spans.iter().map(|s| s.name.as_ref()).collect::<Vec<_>>()
+    );
+}