feat: store Recall history in ClickHouse and add replacement sinks

crates/sqlbuild-core/src/command_output/constants.rs

@@ -9,6 +9,8 @@ pub const MIN_SHUTDOWN_TIMEOUT_SECONDS: f64 = 0.0;
 pub const COMMAND_OUTPUT_RECORD_TYPE: &str = "command_output";
 pub const COMMAND_OUTPUT_LOSS_RECORD_TYPE: &str = "command_output_loss";
 pub const CURRENT_SCHEMA_VERSION: i64 = 1;
+/// The schema of records a Recall replacement sink receives: version 1 plus the exact `data` bytes.
+pub const RECALL_REPLACEMENT_SCHEMA_VERSION: i64 = 2;
 pub const MAX_INVOCATION_CONTEXT_BYTES: usize = 16 * 1024;
 pub const MAX_INVOCATION_CONTEXT_DEPTH: usize = 8;
 pub const INVOCATION_CONTEXT_ENV: &str = "SQLBUILD_INVOCATION_CONTEXT_JSON";

crates/sqlbuild-core/src/command_output/main/ansi_free_text.rs

@@ -0,0 +1,9 @@
+//! `ansi_free_text`: output bytes as the text a command-output record's `message` holds.
+
+use crate::command_output::_helpers::text::strip_ansi;
+
+/// `data` without terminal control sequences, decoded as UTF-8 with invalid bytes replaced.
+#[must_use]
+pub fn ansi_free_text(data: &[u8]) -> String {
+    String::from_utf8_lossy(&strip_ansi(data)).into_owned()
+}

crates/sqlbuild-core/src/command_output/main/mod.rs

@@ -1,3 +1,4 @@
+pub mod ansi_free_text;
 pub mod append_output;
 pub mod chunk_output_text;
 pub mod close_output;

crates/sqlbuild-core/src/command_output/main/validate_output_record.rs

@@ -2,6 +2,7 @@
 
 use crate::command_output::constants::{
     COMMAND_OUTPUT_LOSS_RECORD_TYPE, COMMAND_OUTPUT_RECORD_TYPE, CURRENT_SCHEMA_VERSION,
+    RECALL_REPLACEMENT_SCHEMA_VERSION,
 };
 use crate::command_output::models::RecordShape;
 
@@ -35,6 +36,9 @@ pub fn validate_output_record(shape: &RecordShape) -> Result<(), String> {
     if !shape.message_is_text {
         return Err("message must be a string".to_owned());
     }
+    if !shape.data_is_bytes {
+        return Err("data must be bytes".to_owned());
+    }
     let chunk_index = at_least("chunk_index", shape.chunk_index, 0)?;
     let chunk_count = at_least("chunk_count", shape.chunk_count, 1)?;
     let dropped = at_least("dropped_records", shape.dropped_records, 0)?;
@@ -54,8 +58,13 @@ pub fn validate_output_record(shape: &RecordShape) -> Result<(), String> {
         Some(COMMAND_OUTPUT_RECORD_TYPE | COMMAND_OUTPUT_LOSS_RECORD_TYPE) => {}
         _ => return Err("record_type is unknown".to_owned()),
     }
-    if shape.schema_version != Some(CURRENT_SCHEMA_VERSION) {
-        return Err(format!("schema_version must be {CURRENT_SCHEMA_VERSION}"));
+    if !matches!(
+        shape.schema_version,
+        Some(CURRENT_SCHEMA_VERSION | RECALL_REPLACEMENT_SCHEMA_VERSION)
+    ) {
+        return Err(format!(
+            "schema_version must be {CURRENT_SCHEMA_VERSION} or {RECALL_REPLACEMENT_SCHEMA_VERSION}"
+        ));
     }
     Ok(())
 }

crates/sqlbuild-core/src/command_output/models.rs

@@ -116,6 +116,7 @@ pub struct RecordShape {
     pub occurred_at_is_utc: bool,
     pub stream_is_known: bool,
     pub message_is_text: bool,
+    pub data_is_bytes: bool,
     pub chunk_index: Option<i64>,
     pub chunk_count: Option<i64>,
     pub dropped_records: Option<i64>,

crates/sqlbuild-python/src/bindings/_helpers/cli/entry.rs

@@ -1,5 +1,6 @@
 //! The `sqb` entry: invocation scopes, parsing, observed dispatch and the error projection.
 
+use std::sync::Arc;
 use std::time::Instant;
 
 use pyo3::exceptions::PySystemExit;
@@ -16,7 +17,9 @@ use crate::bindings::_helpers::build_progress::transient_lines::shared_transient
 use crate::bindings::_helpers::cli::argv::parse_cli_invocation;
 use crate::bindings::_helpers::cli::dispatch::{command_facts, dispatch_command};
 use crate::bindings::_helpers::cli::launch_state::offer_launch_record;
-use crate::bindings::_helpers::cli::observability::cli_observability_scope;
+use crate::bindings::_helpers::cli::observability::{
+    CliObservabilityScope, cli_observability_scope,
+};
 use crate::bindings::_helpers::compile_command::report::format_compile_error_json;
 use crate::bindings::_helpers::compile_driver::python_calls::{
     call_with, imported, keywords, within,
@@ -397,12 +400,27 @@ fn lifecycle_event<'py>(
     )
 }
 
-/// Subscribe the invocation's Recall session to every lifecycle fact and SQL artifact.
-fn subscribe_recall(dispatcher: &Bound<'_, PyAny>, recall: &InvocationRecall) -> PyResult<()> {
+/// Install the discovered Recall replacement sinks, release the context row, and subscribe the session to every lifecycle fact and SQL artifact.
+fn subscribe_recall(
+    (scope, dispatcher): (&Bound<'_, CliObservabilityScope>, &Bound<'_, PyAny>),
+    recall: &InvocationRecall,
+) -> PyResult<()> {
     let Some(session) = recall.session.clone() else {
         return Ok(());
     };
-    let sink = Bound::new(dispatcher.py(), RecallLifecycleSink { session })?;
+    let py: Python<'_> = dispatcher.py();
+    let (lifecycle, output) = scope.get().recall_sinks(py)?;
+    let filter_active: bool = recall
+        .replacements
+        .configure(&session, (&lifecycle, &output))?;
+    session.publish_context(filter_active);
+    let sink = Bound::new(
+        py,
+        RecallLifecycleSink {
+            session,
+            replacements: Arc::clone(&recall.replacements),
+        },
+    )?;
     dispatcher.call_method1("subscribe_history", (sink,))?;
     Ok(())
 }
@@ -432,6 +450,9 @@ fn observed_dispatch<'py>(
 ) -> PyResult<Bound<'py, PyAny>> {
     let py: Python<'py> = invocation.py();
     if !policy.observed {
+        if let Some(session) = &invocation.recall.session {
+            session.publish_context(false);
+        }
         return dispatch_command(&invocation.args, invocation.handlers.clone());
     }
     let command: String = invocation.command_name();
@@ -441,20 +462,20 @@ fn observed_dispatch<'py>(
             .unwrap_or(u64::MAX)
             .into_bound_py_any(py)
     };
-    let scope: Bound<'py, PyAny> = Bound::new(
+    let observability: Bound<'py, CliObservabilityScope> = Bound::new(
         py,
         cli_observability_scope(
             invocation.args.clone().unbind(),
             invocation.project_dir()?.unbind(),
         ),
-    )?
-    .into_any();
+    )?;
+    let scope: Bound<'py, PyAny> = observability.clone().into_any();
     within(&scope, |dispatcher| {
         let publish = |event: Bound<'py, PyAny>| -> PyResult<()> {
             dispatcher.call_method1("publish_lifecycle", (event,))?;
             Ok(())
         };
-        subscribe_recall(dispatcher, &invocation.recall)?;
+        subscribe_recall((&observability, dispatcher), &invocation.recall)?;
         let payload: Bound<'py, PyDict> = PyDict::new(py);
         payload.set_item("command", &command)?;
         add_invocation_context(invocation, &payload)?;
@@ -551,7 +572,14 @@ fn parsed_invocation<'py>(
         InvocationRecall::default()
     };
     let recall_sink: Bound<'py, PyAny> = match recall.session.clone() {
-        Some(session) => Bound::new(py, RecallLifecycleSink { session })?.into_any(),
+        Some(session) => Bound::new(
+            py,
+            RecallLifecycleSink {
+                session,
+                replacements: Arc::clone(&recall.replacements),
+            },
+        )?
+        .into_any(),
         None => py.None().into_bound(py),
     };
     let closure_recall: InvocationRecall = recall.clone();

crates/sqlbuild-python/src/bindings/_helpers/cli/observability.rs

@@ -150,6 +150,8 @@ struct Observed {
     exporter_scope: Option<Py<PyAny>>,
     unsubscribe_exporters: Option<Py<PyAny>>,
     unsubscribe_outcome: Option<Py<PyAny>>,
+    /// The discovered Recall lifecycle and command-output replacement sinks, each `(sink, writer_parameter)` or `None`.
+    recall_sinks: Option<(Py<PyAny>, Py<PyAny>)>,
     /// Context managers entered, innermost last.
     entered: Vec<Py<PyAny>>,
 }
@@ -332,6 +334,14 @@ impl Observed {
                 &host::entry(py, UserCodeHost::DiscoverRuntimeExtensions)?,
                 &[("project_dir", project_dir.clone())],
             )?;
+            let lifecycle_replacement: Bound<'_, PyAny> =
+                native_field(&extensions, "recall_lifecycle_sink")?;
+            let output_replacement: Bound<'_, PyAny> =
+                native_field(&extensions, "recall_output_sink")?;
+            if !lifecycle_replacement.is_none() || !output_replacement.is_none() {
+                withhold_launch_record(py)?;
+            }
+            self.recall_sinks = Some((lifecycle_replacement.unbind(), output_replacement.unbind()));
             let event_exporters: Bound<'_, PyAny> = native_field(&extensions, "event_exporters")?;
             let sinks: Bound<'_, PyAny> = native_field(&extensions, "command_output_sinks")?;
             command_output_sinks = sinks.is_truthy()?;
@@ -432,6 +442,23 @@ pub(crate) struct CliObservabilityScope {
 }
 
 impl CliObservabilityScope {
+    /// The Recall lifecycle and command-output replacement sinks discovery found, each `(sink, writer_parameter)` or `None`.
+    pub(crate) fn recall_sinks<'py>(
+        &self,
+        py: Python<'py>,
+    ) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>)> {
+        let observed = self.locked()?;
+        Ok(
+            match observed
+                .as_ref()
+                .and_then(|observed| observed.recall_sinks.as_ref())
+            {
+                Some((lifecycle, output)) => (lifecycle.bind(py).clone(), output.bind(py).clone()),
+                None => (py.None().into_bound(py), py.None().into_bound(py)),
+            },
+        )
+    }
+
     fn locked(&self) -> PyResult<MutexGuard<'_, Option<Observed>>> {
         self.observed
             .lock()
@@ -510,6 +537,7 @@ impl CliObservabilityScope {
             exporter_scope: None,
             unsubscribe_exporters: None,
             unsubscribe_outcome: None,
+            recall_sinks: None,
             entered: Vec::new(),
         };
         if let Err(error) = observed.install(py, (args, self.project_dir.bind(py)), &terminal_index)

crates/sqlbuild-python/src/bindings/_helpers/compile_records/constants.rs

@@ -1191,6 +1191,8 @@ const R_DISCOVERED_RUNTIME_EXTENSIONS: RecordSpec = RecordSpec {
         FieldSpec::new("event_exporters", FieldDefault::EmptyTuple),
         FieldSpec::new("command_output_sinks", FieldDefault::EmptyTuple),
         FieldSpec::new("lifecycle_shutdown_timeout_seconds", FieldDefault::None),
+        FieldSpec::new("recall_lifecycle_sink", FieldDefault::None),
+        FieldSpec::new("recall_output_sink", FieldDefault::None),
     ],
 };
 const R_COMPILER_DIAGNOSTIC: RecordSpec = RecordSpec {

crates/sqlbuild-python/src/bindings/_helpers/discovery/imported_providers.rs

@@ -15,12 +15,15 @@ use crate::bindings::_helpers::discovery::imported_modules::{
 };
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::native_host::user_code::UserCodeHost;
+use crate::bindings::_helpers::recall::writer::RecallWriter;
 use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::fields::mapping_get_or;
 use crate::bindings::_helpers::records::ids::{ErrorId, RecordId};
 
 const PROVIDER_ERROR: ErrorId = ErrorId::ProviderDiscovery;
 const SINK_ERROR: ErrorId = ErrorId::EventExporterDiscovery;
+/// The provider name kept free so a sink's `RecallWriter` parameter can never be mistaken for a provider.
+const RESERVED_PROVIDER_NAME: &str = "recall_writer";
 const PROVIDER_INPUT_ERROR: UserCodeHost = UserCodeHost::ProviderInputError;
 const VALIDATION_ERROR: (&str, &str) = ("pydantic", "ValidationError");
 const SINK_DEFINITION: UserCodeHost = UserCodeHost::GetLifecycleEventSinkDefinition;
@@ -157,6 +160,17 @@ pub(crate) fn discover_provider_classes<'py>(
             let relative_path: Bound<'py, PyAny> = project.relative(&file_path)?;
             let name: Bound<'py, PyAny> = provider_name(&provider_class, &relative_path)?;
             let name_text: String = name.extract()?;
+            if name_text == RESERVED_PROVIDER_NAME {
+                return Err(discovery_error(
+                    py,
+                    PROVIDER_ERROR,
+                    format!(
+                        "Provider name '{RESERVED_PROVIDER_NAME}' in {} is reserved for the \
+                         RecallWriter of Recall replacement sinks; rename the provider",
+                        project.relative_to(&file_path)?
+                    ),
+                ));
+            }
             if let Some((_, existing)) = seen.iter().find(|(listed, _)| *listed == name_text) {
                 return Err(discovery_error(
                     py,
@@ -267,6 +281,33 @@ pub(crate) fn discover_event_exporter_declarations<'py>(
     tuple_of(py, declarations)
 }
 
+/// Each parameter of `function` with its annotation, and whether that annotation is `RecallWriter`.
+fn annotated_parameters<'py>(
+    function: &Bound<'py, PyAny>,
+) -> PyResult<Vec<(String, Bound<'py, PyAny>, bool)>> {
+    let py: Python<'py> = function.py();
+    let writer = py.get_type::<RecallWriter>();
+    let mut parameters: Vec<(String, Bound<'py, PyAny>, bool)> = Vec::new();
+    for pair in shim(py, UserCodeHost::ParameterAnnotations)?
+        .call1((function,))?
+        .try_iter()?
+    {
+        let (parameter, annotation): (String, Bound<'py, PyAny>) = pair?.extract()?;
+        let is_writer: bool = annotation.is(writer.as_any());
+        parameters.push((parameter, annotation, is_writer));
+    }
+    Ok(parameters)
+}
+
+/// Whether `declaration`'s function takes a `RecallWriter`, making it a Recall replacement sink.
+fn takes_recall_writer(declaration: &Bound<'_, PyAny>) -> PyResult<bool> {
+    Ok(
+        annotated_parameters(&native_field(declaration, "function")?)?
+            .iter()
+            .any(|(_, _, is_writer)| *is_writer),
+    )
+}
+
 /// Provider parameters after the event, each naming a known provider of the right class.
 fn exporter_provider_usages<'py>(
     declaration: &Bound<'py, PyAny>,
@@ -280,13 +321,10 @@ fn exporter_provider_usages<'py>(
         .str()?
         .extract()?;
     let empty: Bound<'py, PyAny> = imported(py, "inspect", "Parameter")?.getattr("empty")?;
-    let annotations: Bound<'py, PyAny> =
-        shim(py, UserCodeHost::ParameterAnnotations)?.call1((&function,))?;
-    for (index, pair) in annotations.try_iter()?.enumerate() {
-        if index == 0 {
+    for (parameter, annotation, is_writer) in annotated_parameters(&function)?.into_iter().skip(1) {
+        if is_writer {
             continue;
         }
-        let (parameter, annotation): (String, Bound<'py, PyAny>) = pair?.extract()?;
         let Some(provider) = providers.get_item(&parameter)? else {
             return Err(discovery_error(
                 py,
@@ -344,7 +382,7 @@ pub(crate) fn bind_event_exporter_declarations<'py>(
     tuple_of(py, bound)
 }
 
-/// Validated lifecycle-event sink declarations under sinks/, bound to providers.
+/// Validated lifecycle-event sink declarations under sinks/, bound to providers; Recall replacement sinks are left out.
 #[pyfunction]
 #[pyo3(signature = (*, project_dir, providers = None))]
 pub(crate) fn discover_event_exporter_functions<'py>(
@@ -358,10 +396,15 @@ pub(crate) fn discover_event_exporter_functions<'py>(
     if let Some(cached) = project.cached_extensions()? {
         return cached.get_item(1);
     }
-    let declarations: Bound<'py, PyAny> =
-        discover_event_exporter_declarations(project_dir.clone())?;
+    let mut ordinary: Vec<Bound<'py, PyAny>> = Vec::new();
+    for declaration in discover_event_exporter_declarations(project_dir.clone())?.try_iter()? {
+        let declaration: Bound<'py, PyAny> = declaration?;
+        if !takes_recall_writer(&declaration)? {
+            ordinary.push(declaration);
+        }
+    }
     bind_event_exporter_declarations(
-        declarations,
+        tuple_of(py, ordinary)?,
         providers.map_or_else(|| tuple_of(py, Vec::new()), Ok)?,
         project_dir,
     )

crates/sqlbuild-python/src/bindings/_helpers/native_host/adapter.rs

@@ -44,6 +44,11 @@ pub(crate) enum AdapterHost {
     NormalizeRelationType,
     NormalizeType,
     SafeCastNullability,
+    ConnectRecallStore,
+    RunRecallQuery,
+    RunRecallCommand,
+    InsertRecallRows,
+    CloseRecallStore,
 }
 
 /// Every adapter entry, in `AdapterHost` order.
@@ -243,6 +248,31 @@ pub(crate) const ENTRIES: &[HostEntry] = &[
         kind: HostKind::Function,
         edge: HostEdge::Adapter,
     },
+    HostEntry {
+        name: "connect_recall_store",
+        kind: HostKind::Function,
+        edge: HostEdge::Adapter,
+    },
+    HostEntry {
+        name: "run_recall_query",
+        kind: HostKind::Function,
+        edge: HostEdge::Adapter,
+    },
+    HostEntry {
+        name: "run_recall_command",
+        kind: HostKind::Function,
+        edge: HostEdge::Adapter,
+    },
+    HostEntry {
+        name: "insert_recall_rows",
+        kind: HostKind::Function,
+        edge: HostEdge::Adapter,
+    },
+    HostEntry {
+        name: "close_recall_store",
+        kind: HostKind::Function,
+        edge: HostEdge::Adapter,
+    },
 ];
 
 impl AdapterHost {