refactor: simplify row compute execution facade

Cargo.toml

@@ -165,6 +165,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/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/plan_dry_display.rs

@@ -156,6 +156,9 @@ pub enum PlanDryOp {
     Dedupe {
         keys: Vec<String>,
     },
+    With {
+        columns: Vec<String>,
+    },
     Render {
         columns: Vec<String>,
         template_chars: usize,
@@ -336,6 +339,7 @@ pub(crate) fn human_ux_headline_for_op(op: &PlanDryOp) -> String {
         PlanDryOp::Limit { count } => format!("Take first {count}"),
         PlanDryOp::Dedupe { keys } if keys.is_empty() => "Distinct rows".into(),
         PlanDryOp::Dedupe { keys } => format!("Dedupe on {}", keys.join(", ")),
+        PlanDryOp::With { columns } => format!("Add columns {}", columns.join(", ")),
         PlanDryOp::Render { .. } => "Render text".into(),
         PlanDryOp::ForEach { .. } => "For each row".into(),
         PlanDryOp::Relation { .. } => "Follow relation".into(),
@@ -374,6 +378,7 @@ pub(crate) fn human_ux_summary_for_op(op: &PlanDryOp) -> String {
         PlanDryOp::Aggregate { .. } => "Summarize".into(),
         PlanDryOp::Dedupe { keys } if keys.is_empty() => "Distinct rows".into(),
         PlanDryOp::Dedupe { keys } => format!("Dedupe on {}", keys.join(", ")),
+        PlanDryOp::With { columns } => format!("Add {}", columns.join(", ")),
         PlanDryOp::Render { columns, .. } => format!("Render {}", columns.join(", ")),
         PlanDryOp::Relation {
             relation, target, ..
@@ -408,6 +413,7 @@ pub(crate) fn render_plan_dry_op(op: &PlanDryOp) -> String {
                 format!("dedupe {}", keys.join(", "))
             }
         }
+        PlanDryOp::With { columns } => format!("with {}", columns.join(", ")),
         PlanDryOp::Render {
             columns,
             template_chars,
@@ -499,6 +505,12 @@ fn compact_op_from_compute(
         ComputeOp::DedupeBy { keys } => PlanDryOp::Dedupe {
             keys: keys.iter().map(|k| k.dotted()).collect(),
         },
+        ComputeOp::With { columns } => PlanDryOp::With {
+            columns: columns
+                .iter()
+                .map(|c| c.name.as_str().to_string())
+                .collect(),
+        },
         ComputeOp::Render {
             columns, template, ..
         } => PlanDryOp::Render {

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

@@ -624,7 +624,8 @@ impl<'a, P: FlowPolicyEvaluator + ?Sized> FlowPass<'a, P> {
             ComputeOp::Filter { .. }
             | ComputeOp::Sort { .. }
             | ComputeOp::Limit { .. }
-            | ComputeOp::DedupeBy { .. } => {
+            | ComputeOp::DedupeBy { .. }
+            | ComputeOp::With { .. } => {
                 out = source_facts.clone();
             }
             ComputeOp::GroupBy { aggregates, .. } | ComputeOp::Aggregate { aggregates, .. } => {

crates/plasm-agent-core/src/plasm_dag/postfix/postfix_op.rs

@@ -70,7 +70,26 @@ pub(in crate::plasm_dag) fn postfix_op_to_compute(
                 cgs: cgs.as_ref(),
                 symbol_map: None,
             };
-            plasm_core::type_check_row_predicate(&row_pred, &tc_ctx).map_err(|e| e.to_string())?;
+            let predicates = crate::row_predicate_lower::lower_row_predicate_to_plan(
+                &row_pred,
+                session,
+                &qe,
+                state.cross_cache,
+            )?;
+            let extra = resolve_immediate_compute_schema(state, staged, source);
+            let mut catalog_pred = row_pred.clone();
+            if let Some(schema) = extra.as_ref() {
+                catalog_pred.0.retain(|c| {
+                    !schema
+                        .fields
+                        .iter()
+                        .any(|f| f.name.as_str() == c.field.as_str())
+                });
+            }
+            if !catalog_pred.0.is_empty() {
+                plasm_core::type_check_row_predicate(&catalog_pred, &tc_ctx)
+                    .map_err(|e| e.to_string())?;
+            }
             let mut paths = Vec::new();
             for clause in &row_pred.0 {
                 paths.push(FieldPath::from_dotted(clause.field.as_str())?);
@@ -85,13 +104,13 @@ pub(in crate::plasm_dag) fn postfix_op_to_compute(
                     "filter(...)",
                 )?;
             }
-            let predicates = crate::row_predicate_lower::lower_row_predicate_to_plan(
-                &row_pred,
+            let schema = compute_passthrough_or_fallback_schema(
                 session,
-                &qe,
-                state.cross_cache,
-            )?;
-            let schema = synthetic_schema_passthrough_rows(session, state, staged, source)?;
+                state,
+                staged,
+                source,
+                "PlanFilter",
+            );
             Ok(mk(ComputeOp::Filter { predicates }, schema, false))
         }
         PlasmPostfixOp::Sort { args } => {
@@ -229,10 +248,43 @@ pub(in crate::plasm_dag) fn postfix_op_to_compute(
             let schema = synthetic_schema_passthrough_rows(session, state, staged, source)?;
             Ok(mk(ComputeOp::DedupeBy { keys: vec![] }, schema, false))
         }
+        PlasmPostfixOp::With { body } => {
+            let columns = plasm_core::parse_with_body(body).map_err(|e| e.to_string())?;
+            let schema = synthetic_schema_passthrough_rows(session, state, staged, source)?;
+            let mut schema = schema;
+            for col in &columns {
+                schema.fields.push(plasm_core::SyntheticFieldSchema {
+                    name: col.name.clone(),
+                    value_kind: SyntheticValueKind::Unknown,
+                    source: None,
+                });
+            }
+            Ok(mk(ComputeOp::With { columns }, schema, false))
+        }
         PlasmPostfixOp::Projection { fields } => {
             let qe = resolve_qualified_entity_for_dag_source(state, staged, source.to_string());
+            let source_schema = resolve_immediate_compute_schema(state, staged, source);
             let mut map = BTreeMap::new();
-            for field in parse_field_list(session, state.cross_cache, qe.as_ref(), fields)? {
+            for field in
+                parse_field_list(session, state.cross_cache, qe.as_ref(), fields).or_else(|_| {
+                    fields
+                        .split(',')
+                        .map(|s| s.trim())
+                        .filter(|s| !s.is_empty())
+                        .map(|raw| {
+                            let path = FieldPath::from_dotted(raw)?;
+                            let resolved = resolve_sort_field_path(
+                                session,
+                                state.cross_cache,
+                                qe.as_ref(),
+                                source_schema.as_ref(),
+                                &path,
+                            )?;
+                            Ok(resolved.dotted())
+                        })
+                        .collect::<Result<Vec<String>, String>>()
+                })?
+            {
                 map.insert(
                     OutputName::new(field.clone())?,
                     FieldPath::from_dotted(&field)?,

crates/plasm-agent-core/src/plasm_dag/postfix/row_suffix.rs

@@ -92,6 +92,7 @@ pub(in crate::plasm_dag) fn row_suffix_to_postfix(suffix: &RowSuffix) -> Option<
         RowSuffix::GroupBy { args } => Some(PlasmPostfixOp::GroupBy { args: args.clone() }),
         RowSuffix::Dedupe { keys } => Some(PlasmPostfixOp::Dedupe { keys: keys.clone() }),
         RowSuffix::Distinct { keys } => Some(PlasmPostfixOp::Distinct { keys: keys.clone() }),
+        RowSuffix::With { body } => Some(PlasmPostfixOp::With { body: body.clone() }),
         RowSuffix::Singleton => Some(PlasmPostfixOp::Singleton),
         RowSuffix::PageSize { n } => Some(PlasmPostfixOp::PageSize(*n as usize)),
         RowSuffix::Relation { .. } => None,

crates/plasm-agent-core/src/plasm_dag/schema_validate/compute_schema.rs

@@ -36,21 +36,16 @@ pub(in crate::plasm_dag) fn infer_render_columns_for_node(
                 cols.extend(aggregates.iter().map(|a| a.name.clone()));
                 Ok(cols)
             }
-            ComputeOp::Sort { .. } | ComputeOp::Limit { .. } | ComputeOp::DedupeBy { .. } => {
+            ComputeOp::Sort { .. } | ComputeOp::Limit { .. } | ComputeOp::DedupeBy { .. } | ComputeOp::Filter { .. } => {
                 let parent = lookup_dag_node(state, staged, parent_id.as_str()).ok_or_else(|| {
                     format!("template column inference: missing upstream node `{parent_id}`")
                 })?;
                 infer_render_columns_for_node(session, state, staged, parent)
             }
+            ComputeOp::With { .. } => Ok(schema.fields.iter().map(|f| f.name.clone()).collect()),
             ComputeOp::Render { .. } => Err(
                 "cannot infer columns from a row-to-text template result; bind a row-producing query/relation/projection, or write explicit `[field,...] <<TAG` columns before the template".into(),
             ),
-            ComputeOp::Filter { .. } => {
-                let parent = lookup_dag_node(state, staged, parent_id.as_str()).ok_or_else(|| {
-                    format!("template column inference: missing upstream node `{parent_id}`")
-                })?;
-                infer_render_columns_for_node(session, state, staged, parent)
-            }
         },
         DagNodeSource::Surface {
             qualified_entity, ..

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

@@ -1660,7 +1660,8 @@ fn analyze_static_cardinality(
                     ComputeOp::Project { .. }
                     | ComputeOp::Filter { .. }
                     | ComputeOp::Sort { .. }
-                    | ComputeOp::DedupeBy { .. } => inner(plan, by_id, &compute.source, memo),
+                    | ComputeOp::DedupeBy { .. }
+                    | ComputeOp::With { .. } => inner(plan, by_id, &compute.source, memo),
                     ComputeOp::Limit { count } if *count <= 1 => {
                         CardinalityAnalysis::StaticSingleton
                     }
@@ -1755,7 +1756,8 @@ fn validated_analyze_static_cardinality(
                 ComputeOp::Project { .. }
                 | ComputeOp::Filter { .. }
                 | ComputeOp::Sort { .. }
-                | ComputeOp::DedupeBy { .. } => inner(plan, by_id, c.compute.source.as_str(), memo),
+                | ComputeOp::DedupeBy { .. }
+                | ComputeOp::With { .. } => inner(plan, by_id, c.compute.source.as_str(), memo),
                 ComputeOp::Limit { count } if *count <= 1 => CardinalityAnalysis::StaticSingleton,
                 ComputeOp::Limit { .. } | ComputeOp::GroupBy { .. } => {
                     CardinalityAnalysis::PluralOrUnknown
@@ -1825,6 +1827,11 @@ fn validate_compute_template(
                 validate_predicate(p, node_index, j)?;
             }
         }
+        ComputeOp::With { columns } if columns.is_empty() => {
+            return Err(format!(
+                "plan.nodes[{node_index}].compute.with.columns must be non-empty"
+            ));
+        }
         ComputeOp::GroupBy { aggregates, .. } | ComputeOp::Aggregate { aggregates } => {
             if aggregates.is_empty() {
                 return Err(format!(