Extract param extraction into its own EngineSnapshot::ingest_params stage (ticket 101)

Raw storage, log-stream header derivation, and auto-param detection
were inlined directly in the ingest actor's write closure alongside
metric computation. Pulling them into their own method gives the
ingest arm two clearly separated, independently-callable stages —
param extraction (ingest_params) and metric computation
(apply_all_metrics) — with no behavior change, in preparation for
splitting these onto separate actors (ticket 102).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ

src/engine/handle.rs

@@ -4,7 +4,6 @@
 use crate::config::SettingsHandle;
 use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS};
 use crate::engine::alerts::{self, AlertRuntimeState};
-use crate::engine::data_value::DataValue;
 use crate::engine::index_key::IndexKey;
 use crate::engine::message::EngineMessage;
 use crate::engine::metric_primitive::{MetricPrimitive, Number};
@@ -12,7 +11,6 @@ use crate::engine::snapshot::EngineSnapshot;
 use crate::engine::snapshot::params::log_stream_header_transition;
 use crate::engine::timestamp::Timestamp;
 use crate::engine::{ClearParamsResult, IngestedLine};
-use crate::flavour::LogLiteralHelper;
 use left_right::Absorb;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
@@ -179,21 +177,9 @@ async fn run(
 
                 let text_for_op = Arc::clone(&text);
                 write.mutate(move |snap: &mut EngineSnapshot| {
-                    snap.approx_bytes_total += text_for_op.approx_log_bytes();
-                    snap.index.insert(
-                        IndexKey::IngestedLog {
-                            sequence,
-                            timestamp,
-                        },
-                        DataValue::LogLiteral {
-                            text: Arc::clone(&text_for_op),
-                        },
-                    );
-                    snap.log_count += 1;
-                    if let Some(header) = row_header.as_ref() {
-                        snap.derive_log_stream_params_from_row(header, &text_for_op);
-                    }
-                    snap.auto_declare_params(&text_for_op);
+                    // Two independently-callable stages (ticket 101): primary/param
+                    // extraction, then profile-scoped metric computation.
+                    snap.ingest_params(sequence, timestamp, &text_for_op, row_header.as_ref());
                     snap.apply_all_metrics(sequence, timestamp, &text_for_op);
                     snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op);
                     snap.purge_stale_sort_helper_indexes();