refactor(dynamic): improve SSA receiver type checks, enhance framework bindings, and expand test coverage

src/dynamic/framework/adapters/migration_prisma.rs

@@ -8,6 +8,7 @@
 use crate::dynamic::framework::{FrameworkAdapter, FrameworkBinding};
 use crate::evidence::EntryKind;
 use crate::summary::FuncSummary;
+use crate::summary::ssa_summary::SsaFuncSummary;
 use crate::symbol::Lang;
 
 pub struct MigrationPrismaAdapter;
@@ -44,6 +45,15 @@ fn source_imports_prisma_migration(file_bytes: &[u8]) -> bool {
         .any(|n| file_bytes.windows(n.len()).any(|w| w == *n))
 }
 
+fn name_is_prisma_migration_entry(name: &str) -> bool {
+    matches!(name, "up" | "down" | "migrate" | "deploy" | "seed")
+}
+
+fn typed_container_allows_prisma(container: &str) -> bool {
+    let lc = container.to_ascii_lowercase();
+    lc.contains("prisma")
+}
+
 impl FrameworkAdapter for MigrationPrismaAdapter {
     fn name(&self) -> &'static str {
         ADAPTER_NAME
@@ -56,29 +66,56 @@ impl FrameworkAdapter for MigrationPrismaAdapter {
     fn detect(
         &self,
         summary: &FuncSummary,
-        _ast: tree_sitter::Node<'_>,
+        ast: tree_sitter::Node<'_>,
         file_bytes: &[u8],
     ) -> Option<FrameworkBinding> {
-        let matches_call = super::any_callee_matches(summary, callee_is_prisma_migration);
-        let matches_source = source_imports_prisma_migration(file_bytes);
-        if matches_call || matches_source {
-            Some(FrameworkBinding {
-                adapter: ADAPTER_NAME.to_owned(),
-                kind: EntryKind::Migration { version: None },
-                route: None,
-                request_params: Vec::new(),
-                response_writer: None,
-                middleware: Vec::new(),
-            })
-        } else {
-            None
-        }
+        detect_prisma_migration(summary, None, ast, file_bytes)
     }
+
+    fn detect_with_context(
+        &self,
+        summary: &FuncSummary,
+        ssa_summary: Option<&SsaFuncSummary>,
+        ast: tree_sitter::Node<'_>,
+        file_bytes: &[u8],
+    ) -> Option<FrameworkBinding> {
+        detect_prisma_migration(summary, ssa_summary, ast, file_bytes)
+    }
+}
+
+fn detect_prisma_migration(
+    summary: &FuncSummary,
+    ssa_summary: Option<&SsaFuncSummary>,
+    _ast: tree_sitter::Node<'_>,
+    file_bytes: &[u8],
+) -> Option<FrameworkBinding> {
+    if !source_imports_prisma_migration(file_bytes) {
+        return None;
+    }
+    let raw_call = super::any_callee_matches(summary, callee_is_prisma_migration)
+        && super::typed_receiver_facts_allow(
+            summary,
+            ssa_summary,
+            callee_is_prisma_migration,
+            typed_container_allows_prisma,
+        );
+    if !(name_is_prisma_migration_entry(&summary.name) || raw_call) {
+        return None;
+    }
+    Some(FrameworkBinding {
+        adapter: ADAPTER_NAME.to_owned(),
+        kind: EntryKind::Migration { version: None },
+        route: None,
+        request_params: Vec::new(),
+        response_writer: None,
+        middleware: Vec::new(),
+    })
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::summary::CalleeSite;
 
     fn parse_js(src: &[u8]) -> tree_sitter::Tree {
         let mut parser = tree_sitter::Parser::new();
@@ -102,4 +139,74 @@ mod tests {
         assert_eq!(binding.adapter, "migration-prisma");
         assert!(matches!(binding.kind, EntryKind::Migration { .. }));
     }
+
+    #[test]
+    fn skips_unrelated_helper_in_prisma_file() {
+        let src: &[u8] = b"const { PrismaClient } = require('@prisma/client');\nconst prisma = new PrismaClient();\n\
+            function formatName(name) { return String(name).trim(); }\n\
+            async function up(name) { await prisma.$executeRawUnsafe('CREATE TABLE ' + name); }\n";
+        let tree = parse_js(src);
+        let summary = FuncSummary {
+            name: "formatName".into(),
+            ..Default::default()
+        };
+        assert!(
+            MigrationPrismaAdapter
+                .detect(&summary, tree.root_node(), src)
+                .is_none(),
+            "Prisma import plus migration entry must not bind unrelated helpers",
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_rejects_non_prisma_raw_call() {
+        let src: &[u8] = b"const { PrismaClient } = require('@prisma/client');\n\
+            async function helper(sql) { await cache.$executeRawUnsafe(sql); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "helper".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "cache.$executeRawUnsafe".to_owned(),
+            receiver: Some("cache".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "CacheClient".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            MigrationPrismaAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_keeps_prisma_raw_call() {
+        let src: &[u8] = b"const { PrismaClient } = require('@prisma/client');\n\
+            async function helper(sql) { await prisma.$executeRawUnsafe(sql); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "helper".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "prisma.$executeRawUnsafe".to_owned(),
+            receiver: Some("prisma".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "PrismaClient".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            MigrationPrismaAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_some()
+        );
+    }
 }

src/dynamic/framework/adapters/mod.rs

@@ -407,6 +407,13 @@ fn is_message_middleware_site(callee: &str, text: &str) -> bool {
         || text_lc.contains("validator")
         || text_lc.contains("interceptor")
         || text_lc.contains("middlewarestack")
+        || text_lc.contains("errorhandler")
+        || text_lc.contains("deadletter")
+        || text_lc.contains("dlq")
+        || text_lc.contains("visibilitytimeout")
+        || text_lc.contains("visibility_timeout")
+        || text_lc.contains("queuegroup")
+        || text_lc.contains("queue_group")
 }
 
 fn push_annotation_candidates(lang: Lang, text: &str, out: &mut Vec<MiddlewareShape>) {
@@ -470,7 +477,6 @@ fn is_message_setup_method(name: &str) -> bool {
             | "withValidator"
             | "withMessageValidator"
             | "UseMiddleware"
-            | "QueueSubscribe"
     )
 }
 

src/dynamic/framework/adapters/scheduled_cron.rs

@@ -10,6 +10,7 @@
 use crate::dynamic::framework::{FrameworkAdapter, FrameworkBinding};
 use crate::evidence::EntryKind;
 use crate::summary::FuncSummary;
+use crate::summary::ssa_summary::SsaFuncSummary;
 use crate::symbol::Lang;
 
 pub struct ScheduledCronAdapter;
@@ -65,6 +66,47 @@ fn extract_schedule(file_bytes: &[u8]) -> Option<String> {
     None
 }
 
+fn name_registered_as_cron_job(name: &str, file_bytes: &[u8]) -> bool {
+    if name.is_empty() {
+        return false;
+    }
+    let text = match std::str::from_utf8(file_bytes) {
+        Ok(s) => s,
+        Err(_) => return false,
+    };
+    const SITES: &[&str] = &[
+        "cron.schedule(",
+        "schedule.scheduleJob(",
+        "nodeSchedule.scheduleJob(",
+        "new CronJob(",
+    ];
+    for site in SITES {
+        let mut cursor = 0;
+        while let Some(idx) = text[cursor..].find(site) {
+            let start = cursor + idx + site.len();
+            let rest = &text[start..];
+            let end = rest
+                .find(['\n', ';'])
+                .map(|n| start + n)
+                .unwrap_or_else(|| text.len());
+            let chunk = &text[start..end];
+            if chunk
+                .split(|ch: char| !ch.is_ascii_alphanumeric() && ch != '_' && ch != '$')
+                .any(|part| part == name)
+            {
+                return true;
+            }
+            cursor = end.min(text.len());
+        }
+    }
+    false
+}
+
+fn typed_container_allows_cron(container: &str) -> bool {
+    let lc = container.to_ascii_lowercase();
+    lc.contains("cron") || lc.contains("schedule")
+}
+
 impl FrameworkAdapter for ScheduledCronAdapter {
     fn name(&self) -> &'static str {
         ADAPTER_NAME
@@ -77,31 +119,59 @@ impl FrameworkAdapter for ScheduledCronAdapter {
     fn detect(
         &self,
         summary: &FuncSummary,
-        _ast: tree_sitter::Node<'_>,
+        ast: tree_sitter::Node<'_>,
         file_bytes: &[u8],
     ) -> Option<FrameworkBinding> {
-        let matches_call = super::any_callee_matches(summary, callee_is_cron);
-        let matches_source = source_imports_cron(file_bytes);
-        if matches_call || matches_source {
-            Some(FrameworkBinding {
-                adapter: ADAPTER_NAME.to_owned(),
-                kind: EntryKind::ScheduledJob {
-                    schedule: extract_schedule(file_bytes),
-                },
-                route: None,
-                request_params: Vec::new(),
-                response_writer: None,
-                middleware: Vec::new(),
-            })
-        } else {
-            None
-        }
+        detect_cron(summary, None, ast, file_bytes)
     }
+
+    fn detect_with_context(
+        &self,
+        summary: &FuncSummary,
+        ssa_summary: Option<&SsaFuncSummary>,
+        ast: tree_sitter::Node<'_>,
+        file_bytes: &[u8],
+    ) -> Option<FrameworkBinding> {
+        detect_cron(summary, ssa_summary, ast, file_bytes)
+    }
+}
+
+fn detect_cron(
+    summary: &FuncSummary,
+    ssa_summary: Option<&SsaFuncSummary>,
+    _ast: tree_sitter::Node<'_>,
+    file_bytes: &[u8],
+) -> Option<FrameworkBinding> {
+    if !source_imports_cron(file_bytes) {
+        return None;
+    }
+    let registered = name_registered_as_cron_job(&summary.name, file_bytes);
+    let cron_call = super::any_callee_matches(summary, callee_is_cron)
+        && super::typed_receiver_facts_allow(
+            summary,
+            ssa_summary,
+            callee_is_cron,
+            typed_container_allows_cron,
+        );
+    if !(registered || cron_call) {
+        return None;
+    }
+    Some(FrameworkBinding {
+        adapter: ADAPTER_NAME.to_owned(),
+        kind: EntryKind::ScheduledJob {
+            schedule: extract_schedule(file_bytes),
+        },
+        route: None,
+        request_params: Vec::new(),
+        response_writer: None,
+        middleware: Vec::new(),
+    })
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::summary::CalleeSite;
 
     fn parse_js(src: &[u8]) -> tree_sitter::Tree {
         let mut parser = tree_sitter::Parser::new();
@@ -145,4 +215,75 @@ mod tests {
                 .is_none()
         );
     }
+
+    #[test]
+    fn skips_unregistered_helper_in_cron_file() {
+        let src: &[u8] = b"const cron = require('node-cron');\n\
+            function tick(payload) { console.log(payload); }\n\
+            function formatPayload(payload) { return String(payload); }\n\
+            cron.schedule('*/5 * * * *', tick);\n";
+        let tree = parse_js(src);
+        let summary = FuncSummary {
+            name: "formatPayload".into(),
+            ..Default::default()
+        };
+        assert!(
+            ScheduledCronAdapter
+                .detect(&summary, tree.root_node(), src)
+                .is_none(),
+            "cron import plus a schedule call must not bind unrelated helpers",
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_rejects_non_cron_schedule_call() {
+        let src: &[u8] = b"const cron = require('node-cron');\n\
+            function setup(payload) { queue.schedule(payload); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "setup".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "queue.schedule".to_owned(),
+            receiver: Some("queue".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "TaskQueue".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            ScheduledCronAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_keeps_cron_schedule_call() {
+        let src: &[u8] = b"const cron = require('node-cron');\n\
+            function setup(payload) { cron.schedule('* * * * *', tick); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "setup".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "cron.schedule".to_owned(),
+            receiver: Some("cron".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "NodeCron".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            ScheduledCronAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_some()
+        );
+    }
 }

src/dynamic/framework/adapters/websocket_ws.rs

@@ -7,6 +7,7 @@
 use crate::dynamic::framework::{FrameworkAdapter, FrameworkBinding};
 use crate::evidence::EntryKind;
 use crate::summary::FuncSummary;
+use crate::summary::ssa_summary::SsaFuncSummary;
 use crate::symbol::Lang;
 
 pub struct WebsocketWsAdapter;
@@ -50,6 +51,46 @@ fn extract_path(file_bytes: &[u8]) -> String {
     "/".to_owned()
 }
 
+fn name_registered_as_ws_message_handler(name: &str, file_bytes: &[u8]) -> bool {
+    if name.is_empty() {
+        return false;
+    }
+    let text = match std::str::from_utf8(file_bytes) {
+        Ok(s) => s,
+        Err(_) => return false,
+    };
+    for site in [
+        ".on('message'",
+        ".on(\"message\"",
+        "on('message'",
+        "on(\"message\"",
+    ] {
+        let mut cursor = 0;
+        while let Some(idx) = text[cursor..].find(site) {
+            let start = cursor + idx + site.len();
+            let rest = &text[start..];
+            let end = rest
+                .find(['\n', ';'])
+                .map(|n| start + n)
+                .unwrap_or_else(|| text.len());
+            let chunk = &text[start..end];
+            if chunk
+                .split(|ch: char| !ch.is_ascii_alphanumeric() && ch != '_' && ch != '$')
+                .any(|part| part == name)
+            {
+                return true;
+            }
+            cursor = end.min(text.len());
+        }
+    }
+    false
+}
+
+fn typed_container_allows_ws(container: &str) -> bool {
+    let lc = container.to_ascii_lowercase();
+    lc.contains("websocket") || lc == "ws" || lc == "wss"
+}
+
 impl FrameworkAdapter for WebsocketWsAdapter {
     fn name(&self) -> &'static str {
         ADAPTER_NAME
@@ -62,31 +103,59 @@ impl FrameworkAdapter for WebsocketWsAdapter {
     fn detect(
         &self,
         summary: &FuncSummary,
-        _ast: tree_sitter::Node<'_>,
+        ast: tree_sitter::Node<'_>,
         file_bytes: &[u8],
     ) -> Option<FrameworkBinding> {
-        let matches_call = super::any_callee_matches(summary, callee_is_ws);
-        let matches_source = source_imports_ws(file_bytes);
-        if matches_call || matches_source {
-            Some(FrameworkBinding {
-                adapter: ADAPTER_NAME.to_owned(),
-                kind: EntryKind::WebSocket {
-                    path: extract_path(file_bytes),
-                },
-                route: None,
-                request_params: Vec::new(),
-                response_writer: None,
-                middleware: Vec::new(),
-            })
-        } else {
-            None
-        }
+        detect_ws(summary, None, ast, file_bytes)
     }
+
+    fn detect_with_context(
+        &self,
+        summary: &FuncSummary,
+        ssa_summary: Option<&SsaFuncSummary>,
+        ast: tree_sitter::Node<'_>,
+        file_bytes: &[u8],
+    ) -> Option<FrameworkBinding> {
+        detect_ws(summary, ssa_summary, ast, file_bytes)
+    }
+}
+
+fn detect_ws(
+    summary: &FuncSummary,
+    ssa_summary: Option<&SsaFuncSummary>,
+    _ast: tree_sitter::Node<'_>,
+    file_bytes: &[u8],
+) -> Option<FrameworkBinding> {
+    if !source_imports_ws(file_bytes) {
+        return None;
+    }
+    let registered = name_registered_as_ws_message_handler(&summary.name, file_bytes);
+    let ws_call = super::any_callee_matches(summary, callee_is_ws)
+        && super::typed_receiver_facts_allow(
+            summary,
+            ssa_summary,
+            callee_is_ws,
+            typed_container_allows_ws,
+        );
+    if !(registered || ws_call) {
+        return None;
+    }
+    Some(FrameworkBinding {
+        adapter: ADAPTER_NAME.to_owned(),
+        kind: EntryKind::WebSocket {
+            path: extract_path(file_bytes),
+        },
+        route: None,
+        request_params: Vec::new(),
+        response_writer: None,
+        middleware: Vec::new(),
+    })
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::summary::CalleeSite;
 
     fn parse_js(src: &[u8]) -> tree_sitter::Tree {
         let mut parser = tree_sitter::Parser::new();
@@ -99,7 +168,8 @@ mod tests {
     fn fires_on_ws_server() {
         let src: &[u8] = b"const { WebSocketServer } = require('ws');\n\
             const wss = new WebSocketServer({ port: 0, path: '/feed' });\n\
-            function onMessage(data) { }\n";
+            function onMessage(data) { }\n\
+            wss.on('connection', (socket) => socket.on('message', onMessage));\n";
         let tree = parse_js(src);
         let summary = FuncSummary {
             name: "onMessage".into(),
@@ -113,4 +183,76 @@ mod tests {
             assert_eq!(path, "/feed");
         }
     }
+
+    #[test]
+    fn skips_unregistered_helper_in_ws_file() {
+        let src: &[u8] = b"const { WebSocketServer } = require('ws');\n\
+            const wss = new WebSocketServer({ port: 0, path: '/feed' });\n\
+            function onMessage(data) { }\n\
+            function formatMessage(data) { return String(data); }\n\
+            wss.on('connection', (socket) => socket.on('message', onMessage));\n";
+        let tree = parse_js(src);
+        let summary = FuncSummary {
+            name: "formatMessage".into(),
+            ..Default::default()
+        };
+        assert!(
+            WebsocketWsAdapter
+                .detect(&summary, tree.root_node(), src)
+                .is_none(),
+            "ws import plus a message registration must not bind unrelated helpers",
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_rejects_non_ws_send_call() {
+        let src: &[u8] = b"const { WebSocketServer } = require('ws');\n\
+            function helper(data) { bus.send(data); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "helper".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "bus.send".to_owned(),
+            receiver: Some("bus".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "MessageBus".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            WebsocketWsAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn ssa_receiver_type_keeps_ws_send_call() {
+        let src: &[u8] = b"const { WebSocketServer } = require('ws');\n\
+            function helper(data) { socket.send(data); }\n";
+        let tree = parse_js(src);
+        let mut summary = FuncSummary {
+            name: "helper".into(),
+            ..Default::default()
+        };
+        summary.callees.push(CalleeSite {
+            name: "socket.send".to_owned(),
+            receiver: Some("socket".to_owned()),
+            ordinal: 0,
+            ..Default::default()
+        });
+        let ssa = SsaFuncSummary {
+            typed_call_receivers: vec![(0, "WebSocket".to_owned())],
+            ..Default::default()
+        };
+        assert!(
+            WebsocketWsAdapter
+                .detect_with_context(&summary, Some(&ssa), tree.root_node(), src)
+                .is_some()
+        );
+    }
 }

src/dynamic/framework/auth_markers.rs

@@ -21,7 +21,7 @@
 //!
 //! Distinct from `crate::auth_analysis::auth_markers`, which serves the
 //! static analyser and tracks router auth-gating only (no
-//! CSRF / validation / sanitization / rate-limit categories).  Both
+//! CSRF / validation / sanitization / broker-runtime categories).  Both
 //! modules can grow new entries independently; the static side gates
 //! route-level finding suppression at scan time, this side gates
 //! verifier-side verdict demotion at oracle time.
@@ -56,6 +56,15 @@ pub enum AuthMarkerKind {
     /// Request-rate throttling.  Examples: `rateLimit`,
     /// `ThrottleRequests`, `Rack::Attack`.
     RateLimit,
+    /// Broker error-handler or retry policy.  These preserve useful
+    /// operator context but do not sanitize payload bytes.
+    ErrorHandling,
+    /// Broker dead-letter queue or dead-letter handler.
+    DeadLetterHandling,
+    /// Broker visibility-timeout / lease-extension policy.
+    VisibilityTimeout,
+    /// Broker queue-group / consumer-group delivery guard.
+    QueueGroup,
 }
 
 type ExactRow = (&'static str, AuthMarkerKind);
@@ -105,6 +114,16 @@ const JS_EXACT: &[ExactRow] = &[
     ("expressRateLimit", AuthMarkerKind::RateLimit),
     ("slowDown", AuthMarkerKind::RateLimit),
     ("ThrottlerGuard", AuthMarkerKind::RateLimit),
+    ("errorHandler", AuthMarkerKind::ErrorHandling),
+    ("handleError", AuthMarkerKind::ErrorHandling),
+    ("deadLetterHandler", AuthMarkerKind::DeadLetterHandling),
+    ("deadLetterQueue", AuthMarkerKind::DeadLetterHandling),
+    ("dlq", AuthMarkerKind::DeadLetterHandling),
+    ("visibilityTimeout", AuthMarkerKind::VisibilityTimeout),
+    ("changeMessageVisibility", AuthMarkerKind::VisibilityTimeout),
+    ("queueGroup", AuthMarkerKind::QueueGroup),
+    ("consumerGroup", AuthMarkerKind::QueueGroup),
+    ("groupId", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for Python middleware (Django, Flask, FastAPI,
@@ -141,6 +160,19 @@ const PYTHON_EXACT: &[ExactRow] = &[
     ("RateLimitMiddleware", AuthMarkerKind::RateLimit),
     ("ratelimit", AuthMarkerKind::RateLimit),
     ("throttle", AuthMarkerKind::RateLimit),
+    ("error_handler", AuthMarkerKind::ErrorHandling),
+    ("handle_error", AuthMarkerKind::ErrorHandling),
+    ("dead_letter_handler", AuthMarkerKind::DeadLetterHandling),
+    ("dead_letter_queue", AuthMarkerKind::DeadLetterHandling),
+    ("dlq", AuthMarkerKind::DeadLetterHandling),
+    ("visibility_timeout", AuthMarkerKind::VisibilityTimeout),
+    (
+        "change_message_visibility",
+        AuthMarkerKind::VisibilityTimeout,
+    ),
+    ("queue_group", AuthMarkerKind::QueueGroup),
+    ("consumer_group", AuthMarkerKind::QueueGroup),
+    ("group_id", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for Java middleware (Spring, Quarkus, Micronaut,
@@ -169,6 +201,21 @@ const JAVA_EXACT: &[ExactRow] = &[
         AuthMarkerKind::InputValidation,
     ),
     ("@RateLimited", AuthMarkerKind::RateLimit),
+    ("DefaultErrorHandler", AuthMarkerKind::ErrorHandling),
+    ("CommonErrorHandler", AuthMarkerKind::ErrorHandling),
+    ("ErrorHandler", AuthMarkerKind::ErrorHandling),
+    (
+        "DeadLetterPublishingRecoverer",
+        AuthMarkerKind::DeadLetterHandling,
+    ),
+    ("DeadLetterQueue", AuthMarkerKind::DeadLetterHandling),
+    ("VisibilityTimeout", AuthMarkerKind::VisibilityTimeout),
+    (
+        "ChangeMessageVisibilityRequest",
+        AuthMarkerKind::VisibilityTimeout,
+    ),
+    ("ConsumerGroup", AuthMarkerKind::QueueGroup),
+    ("GroupId", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for PHP middleware (Laravel, Symfony, CodeIgniter).
@@ -191,6 +238,11 @@ const PHP_EXACT: &[ExactRow] = &[
     ("validated", AuthMarkerKind::InputValidation),
     ("throttle", AuthMarkerKind::RateLimit),
     ("ThrottleRequests", AuthMarkerKind::RateLimit),
+    ("error_handler", AuthMarkerKind::ErrorHandling),
+    ("dead_letter_queue", AuthMarkerKind::DeadLetterHandling),
+    ("dlq", AuthMarkerKind::DeadLetterHandling),
+    ("visibility_timeout", AuthMarkerKind::VisibilityTimeout),
+    ("queue_group", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for Ruby middleware (Rails, Sinatra, Hanami, Rack).
@@ -211,6 +263,11 @@ const RUBY_EXACT: &[ExactRow] = &[
     ("validate_params", AuthMarkerKind::InputValidation),
     ("Rack::Attack", AuthMarkerKind::RateLimit),
     ("throttle", AuthMarkerKind::RateLimit),
+    ("error_handler", AuthMarkerKind::ErrorHandling),
+    ("dead_letter_queue", AuthMarkerKind::DeadLetterHandling),
+    ("dlq", AuthMarkerKind::DeadLetterHandling),
+    ("visibility_timeout", AuthMarkerKind::VisibilityTimeout),
+    ("queue_group", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for Go middleware (gin / echo / fiber / chi).
@@ -232,6 +289,15 @@ const GO_EXACT: &[ExactRow] = &[
     ("RateLimit", AuthMarkerKind::RateLimit),
     ("limiter.New", AuthMarkerKind::RateLimit),
     ("middleware.RateLimit", AuthMarkerKind::RateLimit),
+    ("ErrorHandler", AuthMarkerKind::ErrorHandling),
+    ("DeadLetterHandler", AuthMarkerKind::DeadLetterHandling),
+    ("DeadLetterQueue", AuthMarkerKind::DeadLetterHandling),
+    ("DLQ", AuthMarkerKind::DeadLetterHandling),
+    ("ChangeVisibility", AuthMarkerKind::VisibilityTimeout),
+    ("ChangeMessageVisibility", AuthMarkerKind::VisibilityTimeout),
+    ("QueueSubscribe", AuthMarkerKind::QueueGroup),
+    ("QueueGroup", AuthMarkerKind::QueueGroup),
+    ("ConsumerGroup", AuthMarkerKind::QueueGroup),
 ];
 
 /// Exact-name table for Rust middleware (axum / actix / rocket / warp).
@@ -250,6 +316,11 @@ const RUST_EXACT: &[ExactRow] = &[
     ("rate_limit", AuthMarkerKind::RateLimit),
     ("RateLimitLayer", AuthMarkerKind::RateLimit),
     ("tower_governor", AuthMarkerKind::RateLimit),
+    ("error_handler", AuthMarkerKind::ErrorHandling),
+    ("dead_letter_queue", AuthMarkerKind::DeadLetterHandling),
+    ("dlq", AuthMarkerKind::DeadLetterHandling),
+    ("visibility_timeout", AuthMarkerKind::VisibilityTimeout),
+    ("queue_group", AuthMarkerKind::QueueGroup),
 ];
 
 /// Per-language exact-name table dispatch.  Returns the slice that
@@ -544,6 +615,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn broker_runtime_markers_classified_as_non_demoting_context() {
+        assert_eq!(
+            classify(Lang::JavaScript, "visibilityTimeout"),
+            Some(AuthMarkerKind::VisibilityTimeout)
+        );
+        assert_eq!(
+            classify(Lang::Java, "DefaultErrorHandler"),
+            Some(AuthMarkerKind::ErrorHandling)
+        );
+        assert_eq!(
+            classify(Lang::Go, "QueueSubscribe"),
+            Some(AuthMarkerKind::QueueGroup)
+        );
+        assert_eq!(
+            classify(Lang::Python, "dead_letter_queue"),
+            Some(AuthMarkerKind::DeadLetterHandling)
+        );
+    }
+
     #[test]
     fn c_and_cpp_have_no_markers() {
         assert_eq!(classify(Lang::C, "anything"), None);

src/dynamic/middleware_demotion.rs

@@ -12,11 +12,12 @@
 //! [`FrameworkBinding::middleware`] with the names of every middleware /
 //! decorator / interceptor / filter recorded at adapter time.  The
 //! [`crate::dynamic::framework::auth_markers`] registry then classifies
-//! each name into one of six categories.  Only `InputValidation` and
+//! each name into a coarse category.  Only `InputValidation` and
 //! `OutputSanitization` actually mitigate injection sinks: an
 //! authentication check rejects requests without a valid principal but
 //! does not sanitize the request bytes; a CSRF guard does not stop SSRF;
-//! a rate limiter just delays the inevitable.  So the demotion rule is
+//! a rate limiter or broker dead-letter/error/visibility policy changes
+//! delivery semantics but not payload safety.  So the demotion rule is
 //! tight: a `Confirmed`/`ConfirmedProvenOob` verdict whose binding's
 //! middleware vec contains at least one `InputValidation` or
 //! `OutputSanitization` entry is downgraded to
@@ -93,7 +94,8 @@ pub fn is_confirmed_class(verdict: DifferentialVerdict) -> bool {
 /// `InputValidation` and `OutputSanitization` qualify; authentication /
 /// authorization rejects unauthorised callers but does not sanitize the
 /// bytes the caller sends, CSRF protects against cross-origin abuse,
-/// and rate limiting throttles rather than scrubs.
+/// and rate limiting / broker-runtime guards throttle or reroute rather
+/// than scrub.
 fn is_demoting_category(kind: AuthMarkerKind) -> bool {
     matches!(
         kind,
@@ -214,6 +216,21 @@ mod tests {
         assert_eq!(outcome.verdict, DifferentialVerdict::Confirmed);
     }
 
+    #[test]
+    fn broker_runtime_guards_do_not_demote() {
+        let mut outcome = make_outcome(DifferentialVerdict::Confirmed);
+        let binding = make_binding(vec![
+            "visibilityTimeout",
+            "deadLetterQueue",
+            "errorHandler",
+            "queueGroup",
+        ]);
+        let kinds = apply_demotion(&mut outcome, Some(&binding), Lang::JavaScript);
+        assert!(kinds.is_empty());
+        assert_eq!(outcome.verdict, DifferentialVerdict::Confirmed);
+        assert!(outcome.known_guards.is_empty());
+    }
+
     #[test]
     fn input_validation_demotes_confirmed() {
         let mut outcome = make_outcome(DifferentialVerdict::Confirmed);

tests/dynamic_fixtures/websocket/ws/benign.js

@@ -1,5 +1,6 @@
 // Phase 21 — `ws` WebSocket benign control.
 const _NYX_ADAPTER_MARKER = "require('ws')";
+const _NYX_WS_MESSAGE_MARKER = "wss.on('connection', ws => ws.on('message', onMessage))";
 
 function onMessage(data) {
     return 'echoed: ' + JSON.stringify(String(data));

tests/dynamic_fixtures/websocket/ws/vuln.js

@@ -4,6 +4,7 @@
 // WebSocketServer instance.  It splices the message bytes into a
 // child-process command — classic WS → cmdi shape.
 const _NYX_ADAPTER_MARKER = "require('ws')";
+const _NYX_WS_MESSAGE_MARKER = "wss.on('connection', ws => ws.on('message', onMessage))";
 
 const { execSync } = require('child_process');
 

tests/message_handler_corpus.rs

@@ -343,6 +343,14 @@ def handler(envelope):\n    validate_request(envelope)\n",
             "handler",
             &["validateMessage"],
         ),
+        (
+            Lang::JavaScript,
+            b"const { Consumer } = require('sqs-consumer');\n\
+              function handler(env) {}\n\
+              Consumer.create({ queueUrl: 'http://localhost/q', visibilityTimeout: 30, handleMessage: handler });\n",
+            "handler",
+            &["visibilityTimeout"],
+        ),
         (
             Lang::Python,
             b"from google.cloud import pubsub_v1\n\
@@ -379,6 +387,19 @@ def on_message(ch, method, properties, body):\n    validate_request(body)\n",
             "onMessage",
             &["ValidatingMessageConverter"],
         ),
+        (
+            Lang::Java,
+            b"import org.springframework.amqp.rabbit.annotation.RabbitListener;\n\
+              public class Vuln {\n\
+                @RabbitListener(queues = \"work\")\n\
+                public void onMessage(String body) {}\n\
+                public void configure(Factory factory) {\n\
+                  factory.setCommonErrorHandler(new DefaultErrorHandler());\n\
+                }\n\
+              }\n",
+            "onMessage",
+            &["DefaultErrorHandler"],
+        ),
         (
             Lang::Go,
             b"package entry\n\
@@ -386,7 +407,7 @@ def on_message(ch, method, properties, body):\n    validate_request(body)\n",
               func OnMessage(msg *nats.Msg) { ValidatePayload(msg.Data) }\n\
               func init() { nc.QueueSubscribe(\"events\", \"workers\", OnMessage) }\n",
             "OnMessage",
-            &["ValidatePayload"],
+            &["ValidatePayload", "QueueSubscribe"],
         ),
     ];