[pitboss/grind] deferred session-0011 (20260516T052512Z-20f8)

src/dynamic/lang/php.rs

@@ -75,12 +75,17 @@ impl LangEmitter for PhpEmitter {
 
 /// Phase 26 — PHP chain-step harness.
 ///
-/// Emits a `step.php` script that reads `NYX_PREV_OUTPUT` via
-/// `getenv()` and forwards it on stdout.  The PHP probe shim is kept
-/// outside the chain step for now and wired in alongside the Phase 15
-/// emitter follow-up about probe shim splicing.
+/// Splices the PHP probe shim ([`probe_shim`]) in front of a minimal
+/// driver that reads `NYX_PREV_OUTPUT` via `getenv()` and forwards it
+/// on stdout.  The composite re-verifier swaps the trailing forward for
+/// the next member's payload-injection prologue when running a
+/// multi-step chain; the shim has to be in the same file so a chain
+/// step that terminates at a sink can also drive the `__nyx_probe`
+/// channel.
 fn chain_step(prev_output: Option<&[u8]>) -> ChainStepHarness {
-    let source = "<?php\n$prev = getenv(\"NYX_PREV_OUTPUT\");\nif ($prev === false) { $prev = \"\"; }\necho $prev;\n".to_owned();
+    let shim = probe_shim();
+    let driver = "$prev = getenv(\"NYX_PREV_OUTPUT\");\nif ($prev === false) { $prev = \"\"; }\necho $prev;\n";
+    let source = format!("<?php\n{shim}\n{driver}");
     ChainStepHarness {
         source,
         filename: "step.php".to_owned(),
@@ -712,4 +717,27 @@ mod tests {
             "install_crash_guard ordering wrong: payload_pos={payload_pos} install_pos={install_pos} invoke_pos={invoke_pos}",
         );
     }
+
+    #[test]
+    fn chain_step_splices_probe_shim_for_composite_reverify() {
+        let step = chain_step(Some(b"<prev>"));
+        assert!(
+            step.source.contains("__nyx_probe"),
+            "PHP chain step must splice the probe shim"
+        );
+        assert!(
+            step.source.starts_with("<?php"),
+            "PHP chain step must open with <?php"
+        );
+        assert!(
+            step.source.contains("getenv(\"NYX_PREV_OUTPUT\")"),
+            "PHP chain step must keep its NYX_PREV_OUTPUT forwarder"
+        );
+        let shim_pos = step.source.find("__nyx_probe").unwrap();
+        let driver_pos = step.source.find("getenv(\"NYX_PREV_OUTPUT\")").unwrap();
+        assert!(
+            shim_pos < driver_pos,
+            "probe shim must come before the driver so the shim's helpers are in scope when a sink rewrite splices in"
+        );
+    }
 }

src/dynamic/lang/python.rs

@@ -362,6 +362,26 @@ def __nyx_install_crash_guard(sink_callee):
             signal.signal(s, _handler)
         except (OSError, ValueError):
             pass
+
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
 "#
 }
 
@@ -1207,6 +1227,19 @@ mod tests {
         assert!(harness.source.contains("NYX_PROBE_PATH"));
     }
 
+    #[test]
+    fn probe_shim_publishes_stub_sql_recorder() {
+        let shim = probe_shim();
+        assert!(
+            shim.contains("def __nyx_stub_sql_record"),
+            "Python probe shim must define __nyx_stub_sql_record"
+        );
+        assert!(
+            shim.contains("NYX_SQL_LOG"),
+            "stub recorder must read NYX_SQL_LOG"
+        );
+    }
+
     #[test]
     fn shape_detect_flask() {
         let src = "from flask import Flask\napp = Flask(__name__)\n@app.route('/')\ndef index():\n    pass\n";

src/dynamic/lang/ruby.rs

@@ -71,8 +71,16 @@ impl LangEmitter for RubyEmitter {
 }
 
 /// Phase 26 — Ruby chain-step harness.
+///
+/// Splices the Ruby probe shim ([`probe_shim`]) in front of a minimal
+/// driver that reads `NYX_PREV_OUTPUT` from `ENV` and forwards it on
+/// stdout.  Mirrors the Python / Node steps: a step that terminates at
+/// a sink needs the shim in the same file so it can drive the
+/// `__nyx_probe` channel.
 fn chain_step(prev_output: Option<&[u8]>) -> ChainStepHarness {
-    let source = "prev = ENV[\"NYX_PREV_OUTPUT\"] || \"\"\n$stdout.write(prev)\n".to_owned();
+    let shim = probe_shim();
+    let driver = "prev = ENV[\"NYX_PREV_OUTPUT\"] || \"\"\n$stdout.write(prev)\n";
+    let source = format!("{shim}\n{driver}");
     ChainStepHarness {
         source,
         filename: "step.rb".to_owned(),
@@ -768,4 +776,23 @@ mod tests {
             "install_crash_guard ordering wrong: payload_pos={payload_pos} install_pos={install_pos} invoke_pos={invoke_pos}",
         );
     }
+
+    #[test]
+    fn chain_step_splices_probe_shim_for_composite_reverify() {
+        let step = chain_step(Some(b"<prev>"));
+        assert!(
+            step.source.contains("__nyx_probe"),
+            "Ruby chain step must splice the probe shim"
+        );
+        assert!(
+            step.source.contains("ENV[\"NYX_PREV_OUTPUT\"]"),
+            "Ruby chain step must keep its NYX_PREV_OUTPUT forwarder"
+        );
+        let shim_pos = step.source.find("__nyx_probe").unwrap();
+        let driver_pos = step.source.find("ENV[\"NYX_PREV_OUTPUT\"]").unwrap();
+        assert!(
+            shim_pos < driver_pos,
+            "probe shim must come before the driver so a sink rewrite has the shim's helpers in scope"
+        );
+    }
 }

src/dynamic/stubs/mod.rs

@@ -193,6 +193,18 @@ pub trait StubProvider: Send + Sync + std::fmt::Debug {
     /// empty vec (the oracle treats "no events" as "stub was not
     /// touched").
     fn drain_events(&self) -> Vec<StubEvent>;
+
+    /// Optional companion env var that publishes a host-visible
+    /// recording-path the harness can append observations to.  The
+    /// primary [`StubProvider::endpoint`] is the *connection* the
+    /// harness uses (e.g. a SQLite DB path); the recording endpoint is
+    /// the *side channel* a per-language shim helper writes structured
+    /// records into so the host can correlate them on
+    /// [`StubProvider::drain_events`].  Default `None` means the stub
+    /// does not need a side-channel recording path.
+    fn recording_endpoint(&self) -> Option<(&'static str, String)> {
+        None
+    }
 }
 
 /// Aggregate handle the verifier owns for the lifetime of one
@@ -242,11 +254,22 @@ impl StubHarness {
     /// the sandbox env. The order matches `StubHarness::start`'s kinds
     /// argument so later entries override earlier ones if a harness is
     /// re-used with conflicting requests (it currently never is).
+    ///
+    /// Each stub publishes its primary connection endpoint
+    /// ([`StubKind::env_var`]) first, then any companion recording
+    /// endpoint ([`StubProvider::recording_endpoint`]) it owns.  Today
+    /// only [`SqlStub`] publishes a recording endpoint
+    /// (`NYX_SQL_LOG`); the other three stubs keep their primary
+    /// endpoint as the sole pair.
     pub fn endpoints(&self) -> Vec<(&'static str, String)> {
-        self.stubs
-            .iter()
-            .map(|s| (s.kind().env_var(), s.endpoint()))
-            .collect()
+        let mut out = Vec::with_capacity(self.stubs.len() * 2);
+        for s in &self.stubs {
+            out.push((s.kind().env_var(), s.endpoint()));
+            if let Some(pair) = s.recording_endpoint() {
+                out.push(pair);
+            }
+        }
+        out
     }
 
     /// Borrow the underlying stub list (for tests and oracle wiring).
@@ -379,4 +402,29 @@ mod tests {
         assert!(names.contains(&"NYX_HTTP_ENDPOINT"));
         assert!(names.contains(&"NYX_FS_ROOT"));
     }
+
+    #[test]
+    fn endpoints_includes_sql_recording_path_companion_var() {
+        let dir = TempDir::new().unwrap();
+        let h = StubHarness::start(&[StubKind::Sql], dir.path()).unwrap();
+        let pairs = h.endpoints();
+        let names: Vec<&str> = pairs.iter().map(|(n, _)| *n).collect();
+        assert!(
+            names.contains(&"NYX_SQL_ENDPOINT"),
+            "primary endpoint must be present"
+        );
+        assert!(
+            names.contains(&"NYX_SQL_LOG"),
+            "SqlStub recording-path companion env var must be published"
+        );
+        let log_pair = pairs
+            .iter()
+            .find(|(n, _)| *n == "NYX_SQL_LOG")
+            .expect("NYX_SQL_LOG entry");
+        assert!(
+            log_pair.1.ends_with("nyx_sql_stub.queries.log"),
+            "recording path must point at the queries log file, got {}",
+            log_pair.1
+        );
+    }
 }

src/dynamic/stubs/sql.rs

@@ -111,6 +111,11 @@ impl SqlStub {
     }
 }
 
+/// Companion env var that publishes [`SqlStub::log_path`] so a
+/// language-side shim can append executed queries the host will pick
+/// up on [`SqlStub::drain_events`].
+pub const SQL_STUB_LOG_ENV_VAR: &str = "NYX_SQL_LOG";
+
 impl StubProvider for SqlStub {
     fn kind(&self) -> StubKind {
         StubKind::Sql
@@ -120,6 +125,10 @@ impl StubProvider for SqlStub {
         self.db_path.to_string_lossy().into_owned()
     }
 
+    fn recording_endpoint(&self) -> Option<(&'static str, String)> {
+        Some((SQL_STUB_LOG_ENV_VAR, self.log_path.to_string_lossy().into_owned()))
+    }
+
     fn drain_events(&self) -> Vec<StubEvent> {
         let mut cursor = match self.cursor.lock() {
             Ok(g) => g,
@@ -263,4 +272,16 @@ mod tests {
         let stub = SqlStub::start(dir.path()).unwrap();
         assert_eq!(stub.kind(), StubKind::Sql);
     }
+
+    #[test]
+    fn recording_endpoint_publishes_log_path_under_nyx_sql_log() {
+        let dir = TempDir::new().unwrap();
+        let stub = SqlStub::start(dir.path()).unwrap();
+        let pair = stub
+            .recording_endpoint()
+            .expect("SqlStub must publish a recording endpoint");
+        assert_eq!(pair.0, SQL_STUB_LOG_ENV_VAR);
+        assert_eq!(pair.0, "NYX_SQL_LOG");
+        assert_eq!(pair.1, stub.log_path().to_string_lossy());
+    }
 }

tests/dynamic_fixtures/python/async/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 13

tests/dynamic_fixtures/python/celery/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 17

tests/dynamic_fixtures/python/cli/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 14

tests/dynamic_fixtures/python/django/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 15

tests/dynamic_fixtures/python/fastapi/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 16

tests/dynamic_fixtures/python/flask/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 18

tests/dynamic_fixtures/python/generic/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 12

tests/dynamic_fixtures/python/pytest/vuln.py.golden_harness.py

@@ -121,6 +121,26 @@ def __nyx_install_crash_guard(sink_callee):
         except (OSError, ValueError):
             pass
 
+# Phase 10 (Track D.3) stub helpers.  When the verifier spawned a SqlStub it
+# publishes the queries-log path through NYX_SQL_LOG; a sink call site that
+# wants the host-side stub to see its query appends one record-per-call.  The
+# helper is a no-op when NYX_SQL_LOG is unset so the same fixture source still
+# runs under harness modes that didn't spawn a stub.
+def __nyx_stub_sql_record(query, **detail):
+    import os
+    p = os.environ.get("NYX_SQL_LOG")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            for k, v in detail.items():
+                _f.write('# %s: %s\n' % (str(k), str(v)))
+            _f.write(str(query))
+            if not str(query).endswith('\n'):
+                _f.write('\n')
+    except OSError:
+        pass
+
 
 _NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
 _NYX_SINK_LINE = 14

tests/dynamic_fixtures/stubs_e2e/python/sql/vuln/main.py

@@ -0,0 +1,39 @@
+"""Phase 10 (Track D.3) stub-end-to-end fixture: Python + SQL.
+
+The verifier publishes:
+
+* ``NYX_SQL_ENDPOINT`` — absolute path of a SQLite DB the SqlStub owns.
+* ``NYX_SQL_LOG``      — companion log path the harness appends executed
+  queries to so the host SqlStub picks them up on ``drain_events()``.
+
+This fixture exercises both: it opens the stub DB with stdlib ``sqlite3``,
+runs a tautology SELECT (``OR 1=1``), and forwards the executed query to
+the stub through the Python shim helper ``__nyx_stub_sql_record``.  The
+companion test in ``tests/stubs_e2e_per_lang.rs`` splices in
+``crate::dynamic::lang::python::probe_shim`` ahead of this source, runs it
+with both env vars set, and asserts the stub captured the tautology.
+"""
+
+import os
+import sqlite3
+
+
+def main():
+    db_path = os.environ.get("NYX_SQL_ENDPOINT")
+    if not db_path:
+        return
+    query = "SELECT 1 WHERE 'a' = 'a' OR 1=1 --"
+    conn = sqlite3.connect(db_path)
+    try:
+        rows = conn.execute(query).fetchall()
+        for row in rows:
+            print(row[0])
+    finally:
+        conn.close()
+    # Record the executed query through the probe shim so the host
+    # SqlStub captures it on the next drain_events() call.
+    __nyx_stub_sql_record(query, driver="sqlite3")
+
+
+if __name__ == "__main__":
+    main()

tests/stubs_e2e_per_lang.rs

@@ -0,0 +1,144 @@
+//! Phase 10 (Track D.3) — per-(lang, cap) stub end-to-end tests.
+//!
+//! These tests spin up a real boundary stub, splice the per-language
+//! probe shim (which now carries the cap-specific
+//! `__nyx_stub_*_record` helpers) ahead of a fixture's source, run the
+//! resulting program with the stub's endpoint + recording-path env
+//! vars set, then assert the stub captured the boundary event.
+//!
+//! Unlike `tests/stubs_per_cap.rs` (which synthesises harness
+//! behaviour with host-side `SqlStub::record_query` calls), this suite
+//! drives a real interpreter subprocess so the per-language shim
+//! contract is exercised end-to-end.  When the host is missing the
+//! interpreter the test eprintln-skips, matching every other lang
+//! fixture suite in-tree.
+//!
+//! Acceptance bullet from `.pitboss/play/deferred.md` Phase 10
+//! follow-up: the Python+SQL pair is the cheapest first bite —
+//! `sqlite3` is stdlib so no new toolchain dependency is required for
+//! the dynamic CI matrix.
+
+#![cfg(feature = "dynamic")]
+
+use nyx_scanner::dynamic::lang::python::probe_shim as python_probe_shim;
+use nyx_scanner::dynamic::stubs::{SqlStub, StubProvider};
+use std::path::PathBuf;
+use std::process::Command;
+use tempfile::TempDir;
+
+fn python3_available() -> bool {
+    Command::new("python3")
+        .arg("--version")
+        .output()
+        .map(|o| o.status.success())
+        .unwrap_or(false)
+}
+
+fn fixture_path(rel: &str) -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("dynamic_fixtures")
+        .join("stubs_e2e")
+        .join(rel)
+}
+
+#[test]
+fn python_sql_stub_captures_tautology_query_via_shim_recorder() {
+    if !python3_available() {
+        eprintln!("SKIP: python3 not available");
+        return;
+    }
+
+    let workdir = TempDir::new().expect("tempdir");
+    let stub = SqlStub::start(workdir.path()).expect("SqlStub::start");
+
+    // The verifier publishes the SQLite DB path on `NYX_SQL_ENDPOINT`
+    // (primary) and the queries-log path on `NYX_SQL_LOG` (companion).
+    let endpoint = stub.endpoint();
+    let recording = stub
+        .recording_endpoint()
+        .expect("SqlStub must publish a recording endpoint");
+
+    // Splice the probe shim ahead of the fixture source so the
+    // generated program carries the `__nyx_stub_sql_record` helper.
+    // Mirrors the production `PythonEmitter::emit` ordering.
+    let fixture =
+        std::fs::read_to_string(fixture_path("python/sql/vuln/main.py")).expect("read fixture");
+    let mut combined = String::with_capacity(python_probe_shim().len() + fixture.len() + 64);
+    combined.push_str(python_probe_shim());
+    combined.push_str("\n# ── fixture begins ─\n");
+    combined.push_str(&fixture);
+
+    let script_path = workdir.path().join("driver.py");
+    std::fs::write(&script_path, combined).expect("write driver");
+
+    let output = Command::new("python3")
+        .arg(&script_path)
+        .env("NYX_SQL_ENDPOINT", &endpoint)
+        .env(recording.0, &recording.1)
+        .output()
+        .expect("python3 driver");
+    assert!(
+        output.status.success(),
+        "driver must exit 0; stderr = {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    let events = stub.drain_events();
+    assert!(
+        !events.is_empty(),
+        "SqlStub must capture at least one event after the shim recorder fires"
+    );
+    let tautology = events
+        .iter()
+        .find(|e| e.summary.contains("OR 1=1"))
+        .expect("recorded query must contain the tautology marker");
+    assert_eq!(
+        tautology.detail.get("driver").map(String::as_str),
+        Some("sqlite3"),
+        "kwargs passed to __nyx_stub_sql_record must surface as event detail entries"
+    );
+}
+
+#[test]
+fn python_sql_shim_recorder_is_noop_without_log_env() {
+    if !python3_available() {
+        eprintln!("SKIP: python3 not available");
+        return;
+    }
+
+    let workdir = TempDir::new().expect("tempdir");
+    let stub = SqlStub::start(workdir.path()).expect("SqlStub::start");
+
+    // Drive the same fixture but withhold NYX_SQL_LOG.  The shim
+    // helper must be a no-op so the same source still runs cleanly
+    // under harness modes that didn't spawn a stub.
+    let endpoint = stub.endpoint();
+    let fixture =
+        std::fs::read_to_string(fixture_path("python/sql/vuln/main.py")).expect("read fixture");
+    let mut combined = String::new();
+    combined.push_str(python_probe_shim());
+    combined.push('\n');
+    combined.push_str(&fixture);
+    let script_path = workdir.path().join("driver_no_log.py");
+    std::fs::write(&script_path, combined).expect("write driver");
+
+    let output = Command::new("python3")
+        .arg(&script_path)
+        .env("NYX_SQL_ENDPOINT", &endpoint)
+        .env_remove("NYX_SQL_LOG")
+        .output()
+        .expect("python3 driver");
+    assert!(
+        output.status.success(),
+        "driver must exit 0 even without NYX_SQL_LOG; stderr = {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    let events = stub.drain_events();
+    assert!(
+        events.is_empty(),
+        "no events expected when the recording env var is unset, got {} entries",
+        events.len()
+    );
+}