Skip to main content

brokk_bifrost_python/
runtime_values.rs

1//! Bounded syntax evidence for Python process-input keyed reads.
2//!
3//! CPython publishes the process environment as `os.environ` and the launch
4//! argument vector as `sys.argv`. Both are module attributes, so the read's
5//! identity is the import binder plus the lexical scope, never the spelling of
6//! the root name. This module reports what the syntax proves; the reviewed
7//! runtime model in `brokk-bifrost-analysis` owns what the read *means*, and
8//! the executable semantics own the load it joins.
9//!
10//! Every pass is iterative and bounded. An exhausted budget reports an
11//! incomplete result rather than a clean one.
12
13use brokk_bifrost_core::analyzer::Range;
14use brokk_bifrost_core::analyzer::semantic_diagnostics::contains_node as covers;
15use std::collections::{HashMap, HashSet};
16use tree_sitter::Node;
17
18use crate::bindings::{
19    PythonDirectScopeBindingKind, PythonLexicalNameResolution, PythonLexicalScopeInventory,
20    python_direct_scope_bindings_bounded, python_executable_scope_children,
21    python_name_resolution_at,
22};
23use crate::imports::python_import_bindings_from_tree;
24use crate::syntax::{python_parameter_name, python_plain_string_literal};
25
26/// The reviewed process-input containers this syntax pass looks for.
27///
28/// The pair bounds the traversal so an ordinary subscript never becomes a
29/// fact. It is not an identity claim: a read is reported only after the import
30/// binder proves its root names the standard-library module, and only the
31/// activated runtime model decides whether the container is attacker
32/// controlled.
33const PROCESS_INPUT_CONTAINERS: [(&str, &str); 2] = [("os", "environ"), ("sys", "argv")];
34
35/// A statically nameable Python access key.
36///
37/// Dynamic and unsupported forms are retained as facts so a caller preserves
38/// an incomplete answer instead of silently dropping the access.
39#[derive(Clone, Debug, Eq, PartialEq)]
40pub enum PythonRuntimeAccessKey {
41    /// A plain string-literal subscript, such as `os.environ["HOME"]`.
42    Property(String),
43    /// A non-negative integer-literal subscript, such as `sys.argv[1]`.
44    Index(u128),
45    /// The subscript is computed, so the read names no static key.
46    Dynamic,
47    /// The subscript is a slice, a tuple, or another shape whose value is not
48    /// one modeled element.
49    Unsupported,
50}
51
52/// How the root identifier of one access path resolves at its use site.
53#[derive(Clone, Copy, Debug, Eq, PartialEq)]
54pub enum PythonRuntimeRootResolution {
55    /// The module scope binds the root exactly once, through an import of the
56    /// named standard-library module, and no enclosing function rebinds it.
57    ImportedModule,
58    /// A parameter, local, or other binding covers the root at the use site,
59    /// so this occurrence is not the module attribute.
60    LexicallyBound,
61    /// The binder evidence is insufficient to decide the root.
62    Unknown,
63}
64
65/// What this module proves about writes to the reviewed container.
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub enum PythonRuntimeMutationEvidence {
68    NoKnownWrite,
69    KnownWrite,
70    KnownWriteAndUnknownEffects,
71    UnknownEffects,
72}
73
74/// Whether anything that can run before one read could replace the container
75/// or intercept the access.
76#[derive(Clone, Copy, Debug, Eq, PartialEq)]
77pub enum PythonRuntimeAccessorCoverage {
78    NoKnownAccessorEffects,
79    UnknownAccessorEffects,
80}
81
82/// The execution context one read belongs to.
83#[derive(Clone, Copy, Debug, Eq, PartialEq)]
84pub enum PythonRuntimeExecutionContext {
85    Module,
86    DeferredFunction(Range),
87}
88
89/// Why one occurrence of the reviewed path is not a plain keyed read.
90#[derive(Clone, Copy, Debug, Eq, PartialEq)]
91pub enum PythonRuntimeWriteTarget {
92    /// The container itself is assigned or deleted (`sys.argv = [...]`).
93    Container,
94    /// One key of the container is assigned or deleted.
95    Key,
96    /// The container value reaches a position this pass cannot follow: a
97    /// method receiver, a call argument, an alias, or a return.
98    ContainerEscape,
99    /// The module object itself reaches such a position, so an unseen writer
100    /// can replace any of its attributes.
101    ModuleEscape,
102}
103
104/// One occurrence of a reviewed path that is not a plain keyed read.
105#[derive(Clone, Debug, Eq, PartialEq)]
106pub struct PythonRuntimeWrite {
107    /// The absolute module the import binder proved for the root.
108    pub root_name: String,
109    /// The reviewed container this write can reach, when the occurrence names
110    /// one. A module escape reaches every container of its module.
111    pub container: Option<String>,
112    pub range: Range,
113    pub target: PythonRuntimeWriteTarget,
114}
115
116/// One keyed read of a reviewed process-input container.
117#[derive(Clone, Debug, Eq, PartialEq)]
118pub struct PythonRuntimeRead {
119    /// The root identifier as written. Evidence, never an identity claim.
120    pub binding_name: String,
121    /// The absolute module the import binder proved for the root.
122    pub root_name: String,
123    pub container: String,
124    pub access: PythonRuntimeAccessKey,
125    /// The whole subscript expression, where the executable load sits.
126    pub range: Range,
127    pub root_range: Range,
128    /// The `os.environ` attribute node. This is both the container load and
129    /// the structural seed an RQL field-access query addresses, because a
130    /// Python subscript read is anchored on the base of its access chain.
131    pub container_range: Range,
132    pub candidate_anchor: Range,
133    pub key_range: Range,
134    pub lexical_resolution: PythonRuntimeRootResolution,
135    pub mutation: PythonRuntimeMutationEvidence,
136    pub execution: PythonRuntimeExecutionContext,
137    /// Accessor and effect coverage scoped to this read's own execution
138    /// context and ordered against it. A call that cannot run before the read
139    /// does not poison it.
140    pub accessor: PythonRuntimeAccessorCoverage,
141}
142
143/// Everything one bounded pass over one Python module proves.
144#[derive(Clone, Debug, Eq, PartialEq)]
145pub struct PythonRuntimeReadFacts {
146    pub reads: Vec<PythonRuntimeRead>,
147    pub writes: Vec<PythonRuntimeWrite>,
148    /// Number of AST nodes inspected across the bounded passes.
149    pub visited_nodes: usize,
150    /// False only when a bounded traversal could not visit the full tree.
151    /// Dynamic keys, writes, and unknown effects stay represented in rows.
152    pub complete: bool,
153}
154
155/// Shared node budget for the bounded passes.
156#[derive(Debug)]
157struct PythonRuntimeBudget {
158    remaining: usize,
159    exhausted: bool,
160    visited_nodes: usize,
161}
162
163impl PythonRuntimeBudget {
164    /// The node bound keeps a large tree with few matching facts from making
165    /// extraction unbounded. The multiplier mirrors the JavaScript pass.
166    fn for_facts(max_facts: usize) -> Self {
167        Self {
168            remaining: max_facts.saturating_mul(64).max(4096),
169            exhausted: false,
170            visited_nodes: 0,
171        }
172    }
173
174    fn visit(&mut self) -> bool {
175        if self.remaining == 0 {
176            self.exhausted = true;
177            return false;
178        }
179        self.remaining -= 1;
180        self.visited_nodes += 1;
181        true
182    }
183}
184
185/// One import binding that covers a byte offset.
186struct ModuleBinding {
187    local_name: String,
188    /// The top-level package the local name binds.
189    root_module: String,
190    scope_start_byte: usize,
191    scope_end_byte: usize,
192    function_scoped: bool,
193}
194
195/// Report the reviewed process-input keyed reads and the writes that bound
196/// their pristine-input claim.
197///
198/// `max_facts` bounds emitted rows; the derived node bound stops the pass on a
199/// tree large enough to make either traversal unbounded.
200pub fn extract_python_runtime_reads(
201    root: Node<'_>,
202    source: &str,
203    max_facts: usize,
204) -> PythonRuntimeReadFacts {
205    let mut budget = PythonRuntimeBudget::for_facts(max_facts);
206    let bindings = module_bindings(root, source);
207    let mut reads = Vec::new();
208    let mut writes = Vec::new();
209    let mut stack = vec![root];
210    while let Some(node) = stack.pop() {
211        if !budget.visit() {
212            break;
213        }
214        if node.kind() == "attribute"
215            && reads.len() + writes.len() < max_facts
216            && let Some(path) = reviewed_attribute_path(node, source)
217        {
218            let resolution = root_resolution(
219                path.root,
220                &path.root_module,
221                &bindings,
222                root,
223                source,
224                &mut budget,
225            );
226            if resolution == PythonRuntimeRootResolution::ImportedModule {
227                classify_container_occurrence(
228                    node,
229                    &path,
230                    source,
231                    root,
232                    &mut budget,
233                    &mut reads,
234                    &mut writes,
235                );
236            }
237        }
238        if node.kind() == "identifier"
239            && writes.len() + reads.len() < max_facts
240            && let Some(module) =
241                escaping_module_identifier(node, &bindings, root, source, &mut budget)
242        {
243            writes.push(PythonRuntimeWrite {
244                root_name: module,
245                container: None,
246                range: node_source_range(node),
247                target: PythonRuntimeWriteTarget::ModuleEscape,
248            });
249        }
250        let mut cursor = node.walk();
251        stack.extend(node.named_children(&mut cursor));
252    }
253    // A write anywhere in this module refutes the pristine-input claim of
254    // every read of the same container, wherever the read is written.
255    for read in &mut reads {
256        let known_write = writes.iter().any(|write| {
257            write.root_name == read.root_name
258                && write
259                    .container
260                    .as_ref()
261                    .is_none_or(|container| *container == read.container)
262        });
263        read.mutation = match (known_write, read.accessor) {
264            (false, PythonRuntimeAccessorCoverage::NoKnownAccessorEffects) => {
265                PythonRuntimeMutationEvidence::NoKnownWrite
266            }
267            (false, PythonRuntimeAccessorCoverage::UnknownAccessorEffects) => {
268                PythonRuntimeMutationEvidence::UnknownEffects
269            }
270            (true, PythonRuntimeAccessorCoverage::NoKnownAccessorEffects) => {
271                PythonRuntimeMutationEvidence::KnownWrite
272            }
273            (true, PythonRuntimeAccessorCoverage::UnknownAccessorEffects) => {
274                PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
275            }
276        };
277    }
278    reads.sort_by_key(|read| read.range.start_byte);
279    writes.sort_by_key(|write| write.range.start_byte);
280    PythonRuntimeReadFacts {
281        complete: !budget.exhausted && reads.len() + writes.len() < max_facts,
282        visited_nodes: budget.visited_nodes,
283        reads,
284        writes,
285    }
286}
287
288/// The structural node that represents one reviewed keyed read.
289///
290/// A Python subscript expression has no structural fact of its own, so the
291/// read is addressed by the base of its access chain: the
292/// `<module>.<container>` attribute. This is the same seed the extractor
293/// publishes as `candidate_anchor`, so a query and the executable load agree
294/// on one identity. A node that is not a reviewed keyed read has no seed.
295pub fn python_runtime_keyed_access_seed<'tree>(node: Node<'tree>) -> Option<Node<'tree>> {
296    if node.kind() != "subscript" {
297        return None;
298    }
299    let base = node.child_by_field_name("value")?;
300    if base.kind() != "attribute" {
301        return None;
302    }
303    let attribute = base.child_by_field_name("attribute")?;
304    let object = base.child_by_field_name("object")?;
305    if object.kind() != "identifier" || attribute.kind() != "identifier" {
306        return None;
307    }
308    Some(base)
309}
310
311/// One reviewed `<module>.<container>` attribute occurrence.
312struct ReviewedPath<'tree> {
313    root: Node<'tree>,
314    root_module: String,
315    container: String,
316}
317
318/// Recognize `<identifier>.<container>` for one reviewed pair. Deeper paths
319/// (`pkg.os.environ`) have a non-identifier object and are not this module.
320fn reviewed_attribute_path<'tree>(node: Node<'tree>, source: &str) -> Option<ReviewedPath<'tree>> {
321    let object = node.child_by_field_name("object")?;
322    let attribute = node.child_by_field_name("attribute")?;
323    if object.kind() != "identifier" || attribute.kind() != "identifier" {
324        return None;
325    }
326    let container = node_text(attribute, source);
327    let (root_module, _) = PROCESS_INPUT_CONTAINERS
328        .iter()
329        .find(|(_, reviewed)| *reviewed == container)?;
330    Some(ReviewedPath {
331        root: object,
332        root_module: (*root_module).to_owned(),
333        container: container.to_owned(),
334    })
335}
336
337/// Decide whether one reviewed container occurrence is a keyed read, a write,
338/// or an escape, and record the matching row.
339fn classify_container_occurrence<'tree>(
340    attribute: Node<'tree>,
341    path: &ReviewedPath<'tree>,
342    source: &str,
343    module_root: Node<'tree>,
344    budget: &mut PythonRuntimeBudget,
345    reads: &mut Vec<PythonRuntimeRead>,
346    writes: &mut Vec<PythonRuntimeWrite>,
347) {
348    let write = |target| PythonRuntimeWrite {
349        root_name: path.root_module.clone(),
350        container: Some(path.container.clone()),
351        range: node_source_range(attribute),
352        target,
353    };
354    let Some(parent) = attribute.parent() else {
355        writes.push(write(PythonRuntimeWriteTarget::ContainerEscape));
356        return;
357    };
358    if parent.kind() != "subscript"
359        || parent
360            .child_by_field_name("value")
361            .is_none_or(|value| value.id() != attribute.id())
362    {
363        // Every other position -- a method receiver such as
364        // `os.environ.update(...)`, an argument, an alias, an assignment
365        // target, a `del` -- can replace or observe the container, so the
366        // pristine-input claim of every read of it is refuted.
367        writes.push(write(if is_binding_target(attribute) {
368            PythonRuntimeWriteTarget::Container
369        } else {
370            PythonRuntimeWriteTarget::ContainerEscape
371        }));
372        return;
373    }
374    if is_binding_target(parent) {
375        writes.push(write(PythonRuntimeWriteTarget::Key));
376        return;
377    }
378    let Some(subscript) = parent.child_by_field_name("subscript") else {
379        writes.push(write(PythonRuntimeWriteTarget::ContainerEscape));
380        return;
381    };
382    let access = access_key(subscript, source);
383    let read_range = node_source_range(parent);
384    let execution = match enclosing_execution_context(attribute) {
385        Some(function) => {
386            PythonRuntimeExecutionContext::DeferredFunction(node_source_range(function))
387        }
388        None => PythonRuntimeExecutionContext::Module,
389    };
390    let accessor = execution_context_accessor_coverage(attribute, module_root, read_range, budget);
391    reads.push(PythonRuntimeRead {
392        binding_name: node_text(path.root, source).to_owned(),
393        root_name: path.root_module.clone(),
394        container: path.container.clone(),
395        access,
396        range: read_range,
397        root_range: node_source_range(path.root),
398        container_range: node_source_range(attribute),
399        candidate_anchor: node_source_range(attribute),
400        key_range: node_source_range(subscript),
401        lexical_resolution: PythonRuntimeRootResolution::ImportedModule,
402        // Replaced by the module-wide join once every write is known.
403        mutation: PythonRuntimeMutationEvidence::UnknownEffects,
404        execution,
405        accessor,
406    });
407}
408
409/// The static key one subscript names.
410fn access_key(subscript: Node<'_>, source: &str) -> PythonRuntimeAccessKey {
411    if let Some(literal) = python_plain_string_literal(subscript, source) {
412        return PythonRuntimeAccessKey::Property(literal.to_owned());
413    }
414    match subscript.kind() {
415        "integer" => node_text(subscript, source)
416            .parse::<u128>()
417            .map_or(PythonRuntimeAccessKey::Unsupported, |index| {
418                PythonRuntimeAccessKey::Index(index)
419            }),
420        "slice" | "tuple" | "list_splat" => PythonRuntimeAccessKey::Unsupported,
421        "string" | "concatenated_string" => PythonRuntimeAccessKey::Unsupported,
422        _ => PythonRuntimeAccessKey::Dynamic,
423    }
424}
425
426/// Whether this node is written rather than read: an assignment target, an
427/// augmented-assignment target, a `del` operand, or a `for` target.
428fn is_binding_target(node: Node<'_>) -> bool {
429    let mut current = node;
430    while let Some(parent) = current.parent() {
431        match parent.kind() {
432            "assignment" | "augmented_assignment" | "for_statement" | "for_in_clause" => {
433                return parent
434                    .child_by_field_name("left")
435                    .is_some_and(|left| left.id() == current.id());
436            }
437            "delete_statement" => return true,
438            "pattern_list" | "tuple_pattern" | "list_pattern" | "list_splat_pattern" => {
439                current = parent;
440            }
441            _ => return false,
442        }
443    }
444    false
445}
446
447/// The innermost function or lambda whose body executes this node, or `None`
448/// when the node is outside a function or lambda body.
449pub(crate) fn enclosing_execution_context<'tree>(node: Node<'tree>) -> Option<Node<'tree>> {
450    let mut current = node;
451    while let Some(parent) = current.parent() {
452        if matches!(parent.kind(), "function_definition" | "lambda")
453            && parent
454                .child_by_field_name("body")
455                .is_some_and(|body| covers(body, current))
456        {
457            return Some(parent);
458        }
459        current = parent;
460    }
461    None
462}
463
464/// Whether anything that can run before `read_range` inside this read's own
465/// execution context could replace the container or intercept the access.
466///
467/// Every call is treated as such a hazard. Python has no syntax that proves a
468/// callee cannot reach `os.environ`, so a call the read cannot be ordered
469/// after leaves the read typed incomplete rather than clean.
470fn execution_context_accessor_coverage(
471    attribute: Node<'_>,
472    module_root: Node<'_>,
473    read_range: Range,
474    budget: &mut PythonRuntimeBudget,
475) -> PythonRuntimeAccessorCoverage {
476    let scope = enclosing_execution_context(attribute)
477        .and_then(|function| function.child_by_field_name("body"))
478        .unwrap_or(module_root);
479    let mut stack = vec![scope];
480    while let Some(node) = stack.pop() {
481        if !budget.visit() {
482            return PythonRuntimeAccessorCoverage::UnknownAccessorEffects;
483        }
484        if matches!(node.kind(), "function_definition" | "lambda") {
485            // A nested body is a separate execution context. Its header still
486            // runs here, so only the body is skipped.
487            let mut cursor = node.walk();
488            stack.extend(node.named_children(&mut cursor).filter(|child| {
489                Some(child.id()) != node.child_by_field_name("body").map(|body| body.id())
490            }));
491            continue;
492        }
493        if node.kind() == "call" && can_precede(node_source_range(node), read_range, node) {
494            return PythonRuntimeAccessorCoverage::UnknownAccessorEffects;
495        }
496        let mut cursor = node.walk();
497        stack.extend(node.named_children(&mut cursor));
498    }
499    PythonRuntimeAccessorCoverage::NoKnownAccessorEffects
500}
501
502/// Whether `hazard` can execute before a read at `read_range`.
503///
504/// Source order decides within one straight-line context. A call that ends
505/// before the read starts has run before it. A call that encloses the read,
506/// such as `os.system(os.environ["K"])`, has not: its callee and arguments are
507/// evaluated first, and the call itself runs after the read's value exists.
508/// A hazard inside a loop that also contains the read can run before it on a
509/// later iteration, so the loop is checked structurally rather than by
510/// position alone.
511fn can_precede(hazard: Range, read_range: Range, hazard_node: Node<'_>) -> bool {
512    if hazard.end_byte <= read_range.start_byte {
513        return true;
514    }
515    let mut current = hazard_node;
516    while let Some(parent) = current.parent() {
517        if matches!(parent.kind(), "for_statement" | "while_statement")
518            && parent.start_byte() <= read_range.start_byte
519            && read_range.end_byte <= parent.end_byte()
520        {
521            return true;
522        }
523        current = parent;
524    }
525    false
526}
527
528/// The module whose object escapes at this identifier, when one does.
529///
530/// A module object is shared process-wide, so any position other than the
531/// object of an attribute access can carry it to a writer this pass cannot
532/// see. Its own import statement is the binder, not a use.
533fn escaping_module_identifier(
534    node: Node<'_>,
535    bindings: &[ModuleBinding],
536    module_root: Node<'_>,
537    source: &str,
538    budget: &mut PythonRuntimeBudget,
539) -> Option<String> {
540    let name = node_text(node, source);
541    let module = binding_root_module(node, name, bindings)?;
542    // A parameter or local that happens to spell the module name is not the
543    // module object, so it cannot carry it anywhere. The same three facts that
544    // decide a read decide this.
545    if root_resolution(node, &module, bindings, module_root, source, budget)
546        != PythonRuntimeRootResolution::ImportedModule
547    {
548        return None;
549    }
550    let parent = node.parent()?;
551    if parent.kind() == "attribute"
552        && parent
553            .child_by_field_name("object")
554            .is_some_and(|object| object.id() == node.id())
555    {
556        return None;
557    }
558    if matches!(
559        parent.kind(),
560        "import_statement" | "import_from_statement" | "aliased_import" | "dotted_name"
561    ) {
562        return None;
563    }
564    Some(module)
565}
566
567/// The top-level module one in-scope import binding gives `name`, when every
568/// in-scope binding of the name agrees on it.
569fn binding_root_module(
570    reference: Node<'_>,
571    name: &str,
572    bindings: &[ModuleBinding],
573) -> Option<String> {
574    let offset = reference.start_byte();
575    let mut module = None;
576    for binding in bindings {
577        if binding.local_name != name
578            || offset < binding.scope_start_byte
579            || offset > binding.scope_end_byte
580        {
581            continue;
582        }
583        match &module {
584            None => module = Some(binding.root_module.clone()),
585            Some(existing) if *existing == binding.root_module => {}
586            Some(_) => return None,
587        }
588    }
589    module.filter(|module| {
590        PROCESS_INPUT_CONTAINERS
591            .iter()
592            .any(|(reviewed, _)| reviewed == module)
593    })
594}
595
596/// Resolve the root identifier of one reviewed path.
597///
598/// Three independent structured facts have to agree before an occurrence is
599/// the module attribute: no enclosing function binds the name, the module
600/// scope binds it through imports alone, and every in-scope import binding
601/// names the same reviewed package.
602fn root_resolution(
603    root: Node<'_>,
604    expected_module: &str,
605    bindings: &[ModuleBinding],
606    module_root: Node<'_>,
607    source: &str,
608    budget: &mut PythonRuntimeBudget,
609) -> PythonRuntimeRootResolution {
610    let name = node_text(root, source);
611    match python_name_resolution_at(name, root, source) {
612        PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal => {
613            // A function-scoped import binds the name locally and still names
614            // the module. This pass does not carry that proof, so the root
615            // stays undecided rather than conclusively excluded.
616            return if bindings
617                .iter()
618                .any(|binding| binding.local_name == name && binding.function_scoped)
619            {
620                PythonRuntimeRootResolution::Unknown
621            } else {
622                PythonRuntimeRootResolution::LexicallyBound
623            };
624        }
625        PythonLexicalNameResolution::Global | PythonLexicalNameResolution::Unbound => {}
626    }
627    match module_scope_binds_name_through_imports(module_root, name, source, budget) {
628        Some(true) => {}
629        Some(false) => return PythonRuntimeRootResolution::LexicallyBound,
630        None => return PythonRuntimeRootResolution::Unknown,
631    }
632    match binding_root_module(root, name, bindings) {
633        Some(module) if module == expected_module => PythonRuntimeRootResolution::ImportedModule,
634        Some(_) => PythonRuntimeRootResolution::LexicallyBound,
635        None => PythonRuntimeRootResolution::Unknown,
636    }
637}
638
639/// Whether the module scope binds `name`, and binds it through imports alone.
640///
641/// `Some(false)` says something other than an import binds the name, so the
642/// occurrence is not the module object. `None` says the bounded walk could not
643/// decide, which is not an exclusion.
644fn module_scope_binds_name_through_imports(
645    module_root: Node<'_>,
646    name: &str,
647    source: &str,
648    budget: &mut PythonRuntimeBudget,
649) -> Option<bool> {
650    let mut imported = false;
651    let mut stack = vec![module_root];
652    while let Some(node) = stack.pop() {
653        if !budget.visit() {
654            return None;
655        }
656        if node.kind() == "wildcard_import" {
657            return None;
658        }
659        for binding in python_direct_scope_bindings_bounded(node, source, || budget.visit())? {
660            if node_text(binding.declaration, source) != name {
661                continue;
662            }
663            match binding.kind {
664                PythonDirectScopeBindingKind::Import => imported = true,
665                _ => return Some(false),
666            }
667        }
668        // A nested callable or class body is a separate scope; its header
669        // still executes here.
670        stack.extend(python_executable_scope_children(node));
671    }
672    Some(imported)
673}
674
675/// The import bindings of this module that can name a reviewed module, with
676/// every name the module scope binds more than once removed.
677///
678/// `import os` and `import os.path` both bind the package `os`, so they agree.
679/// A later `os = ...`, a `def os`, or a `from x import os` does not, and the
680/// name then names no proven module.
681fn module_bindings(root: Node<'_>, source: &str) -> Vec<ModuleBinding> {
682    let mut bindings = Vec::new();
683    for binding in python_import_bindings_from_tree(root, source) {
684        let Some(root_module) = binding.qualified_name.split('.').next() else {
685            continue;
686        };
687        if !PROCESS_INPUT_CONTAINERS
688            .iter()
689            .any(|(reviewed, _)| *reviewed == root_module)
690        {
691            continue;
692        }
693        // A `from os import environ` binds the member, not the module object.
694        if binding.qualified_name != root_module && binding.consumed_attributes == 0 {
695            continue;
696        }
697        bindings.push(ModuleBinding {
698            local_name: binding.local_name.clone(),
699            root_module: root_module.to_owned(),
700            scope_start_byte: binding.scope_start_byte,
701            scope_end_byte: binding.scope_end_byte,
702            function_scoped: binding.is_function_scoped(),
703        });
704    }
705    bindings
706}
707
708use brokk_bifrost_core::analyzer::usages::reference_site::node_range as node_source_range;
709
710fn node_text<'a>(node: Node<'_>, source: &'a str) -> &'a str {
711    brokk_bifrost_core::analyzer::common::try_node_source_text(node, source)
712        .expect("a tree-sitter node range is valid UTF-8 source")
713}
714
715/// Why one Python module's member value may not be the module's own at a call.
716#[derive(Clone, Debug, Eq, PartialEq)]
717pub enum PythonModuleMemberMutation {
718    /// A write-shaped target names the member, or names it dynamically.
719    Write(Range),
720    /// The module object reaches a position this file-local walk cannot
721    /// follow, so an unseen writer can replace any of its members.
722    Escape(Range),
723    /// The bounded walk did not finish, so this file proves nothing.
724    Budget,
725}
726
727/// What in this file can replace `module.member`, or `None` when nothing can.
728///
729/// An imported module object is shared process-wide, and Python permits
730/// replacing one of its attributes (`os.system = fake`). A reviewed summary
731/// describes the standard-library implementation, not a value written over it,
732/// so a consumer that closes a call on that summary must first prove no
733/// workspace source can perform the replacement. Two evidence families refuse
734/// it, and both fail closed:
735///
736///  1. A write-shaped target -- an assignment, an augmented assignment, or a
737///     `del` -- whose attribute access resolves to the module. A statically
738///     named attribute refuses that member; `setattr` and every other dynamic
739///     form arrives through the escape rule below.
740///  2. Any use of a name bound to the module outside its own import statement
741///     and its attribute-object position. An alias, a call argument, a return,
742///     or a subscript can carry the shared module object to a writer this walk
743///     cannot see, so the proof refuses every member instead of following the
744///     value. A `sys.modules` read is classified by its structured key;
745///     dynamic-key results are possible aliases and use the same alias
746///     handling. A table write matters only when its stored value is or may
747///     be this module.
748///
749/// A file that never binds the module proves the absence vacuously. The walk
750/// is iterative and bounded; exhausting the bound refuses every member.
751pub fn python_module_member_mutation(
752    root: Node<'_>,
753    source: &str,
754    source_module: &str,
755    module: &str,
756    member: &str,
757    max_nodes: usize,
758) -> Option<PythonModuleMemberMutation> {
759    let mut budget = PythonRuntimeBudget {
760        remaining: max_nodes,
761        exhausted: false,
762        visited_nodes: 0,
763    };
764    let bindings = python_import_bindings_from_tree(root, source);
765    let facts =
766        python_sys_modules_aliases(root, source, source_module, module, &bindings, &mut budget)?;
767    let table_aliases = &facts.aliases;
768    let locals: Vec<String> = bindings
769        .iter()
770        .filter(|binding| {
771            binding.qualified_name == module
772                || binding
773                    .qualified_name
774                    .split('.')
775                    .next()
776                    .is_some_and(|root_module| {
777                        root_module == module && binding.consumed_attributes > 0
778                    })
779        })
780        .map(|binding| binding.local_name.clone())
781        .collect();
782    let mut stack = vec![root];
783    while let Some(node) = stack.pop() {
784        if !budget.visit() {
785            return Some(PythonModuleMemberMutation::Budget);
786        }
787        if node.kind() == "call"
788            && python_mock_patch_may_replace_module_member(node, source, &bindings, module, member)
789        {
790            return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
791        }
792        if node.kind() == "identifier" {
793            let name = node_text(node, source);
794            if let Some(table) = python_sys_modules_table(node, source) {
795                if python_imported_path(table, source, &bindings)
796                    .is_some_and(|path| path == ["sys", "modules"])
797                {
798                    match python_sys_modules_lookup(
799                        table,
800                        source,
801                        source_module,
802                        module,
803                        &bindings,
804                        &facts,
805                        &mut budget,
806                    ) {
807                        PythonSysModulesLookup::Unrelated => {}
808                        PythonSysModulesLookup::Escape(range) => {
809                            return Some(PythonModuleMemberMutation::Escape(range));
810                        }
811                        PythonSysModulesLookup::Target(read) => {
812                            if let Some(mutation) =
813                                python_sys_modules_read_use(read, source, member, table_aliases)
814                            {
815                                return Some(mutation);
816                            }
817                        }
818                    }
819                } else {
820                    return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
821                }
822            }
823            let table_alias = table_aliases.contains(name);
824            if table_alias && python_module_table_alias_binding_use(node, source, table_aliases) {
825                // A recognized table lookup binding is followed at its uses.
826            } else if (table_alias
827                || (locals.iter().any(|local| local == name)
828                    && !matches!(
829                        python_name_resolution_at(name, node, source),
830                        PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
831                    )))
832                && let Some(mutation) =
833                    module_binding_use(node, source, &bindings, table_aliases, module, member)
834            {
835                return Some(mutation);
836            }
837        }
838        let mut cursor = node.walk();
839        stack.extend(node.named_children(&mut cursor));
840    }
841    None
842}
843
844enum PythonSysModulesLookup<'tree> {
845    Unrelated,
846    Target(Node<'tree>),
847    Escape(Range),
848}
849
850/// Return the `sys.modules` attribute when an identifier is its object.
851fn python_sys_modules_table<'tree>(node: Node<'tree>, source: &str) -> Option<Node<'tree>> {
852    let table = node.parent()?;
853    (table.kind() == "attribute"
854        && table
855            .child_by_field_name("object")
856            .is_some_and(|object| object.id() == node.id())
857        && table
858            .child_by_field_name("attribute")
859            .is_some_and(|attribute| node_text(attribute, source) == "modules"))
860    .then_some(table)
861}
862
863fn python_sys_modules_read_table<'tree>(read: Node<'tree>, source: &str) -> Option<Node<'tree>> {
864    if read.kind() == "subscript" {
865        let table = read.child_by_field_name("value")?;
866        return (table.kind() == "attribute"
867            && table
868                .child_by_field_name("attribute")
869                .is_some_and(|attribute| node_text(attribute, source) == "modules"))
870        .then_some(table);
871    }
872    if read.kind() == "call" {
873        let function = read.child_by_field_name("function")?;
874        if function.kind() != "attribute"
875            || function
876                .child_by_field_name("attribute")
877                .is_none_or(|attribute| {
878                    !matches!(node_text(attribute, source), "get" | "pop" | "setdefault")
879                })
880        {
881            return None;
882        }
883        let table = function.child_by_field_name("object")?;
884        return (table.kind() == "attribute"
885            && table
886                .child_by_field_name("attribute")
887                .is_some_and(|attribute| node_text(attribute, source) == "modules"))
888        .then_some(table);
889    }
890    None
891}
892
893fn python_sys_modules_lookup<'tree>(
894    table: Node<'tree>,
895    source: &str,
896    source_module: &str,
897    module: &str,
898    bindings: &[crate::imports::PythonImportBinding],
899    facts: &PythonSysModulesFacts<'tree>,
900    budget: &mut PythonRuntimeBudget,
901) -> PythonSysModulesLookup<'tree> {
902    // The registry maps names to module objects; changing one entry does not
903    // mutate the referenced module's globals. Only storing or producing this
904    // queried module as a value can create another write-through alias.
905    let escaped = || PythonSysModulesLookup::Escape(node_source_range(table));
906    if python_imported_path(table, source, bindings).is_none_or(|path| path != ["sys", "modules"]) {
907        return escaped();
908    }
909    let Some(parent) = table.parent() else {
910        return escaped();
911    };
912    let proof = PythonSysModulesProofContext {
913        source,
914        source_module,
915        module,
916        bindings,
917        facts,
918    };
919    if parent.kind() == "subscript"
920        && parent
921            .child_by_field_name("value")
922            .is_some_and(|value| value.id() == table.id())
923    {
924        if is_binding_target(parent) {
925            if parent
926                .parent()
927                .is_some_and(|parent| parent.kind() == "delete_statement")
928            {
929                return PythonSysModulesLookup::Unrelated;
930            }
931            let Some(value) = parent
932                .parent()
933                .filter(|assignment| assignment.kind() == "assignment")
934                .and_then(|assignment| assignment.child_by_field_name("right"))
935            else {
936                return escaped();
937            };
938            return match python_sys_modules_stored_value(
939                value,
940                &proof,
941                budget,
942                &mut HashSet::default(),
943            ) {
944                PythonSysModulesStoredValue::Unrelated => PythonSysModulesLookup::Unrelated,
945                PythonSysModulesStoredValue::Target | PythonSysModulesStoredValue::Unknown => {
946                    escaped()
947                }
948            };
949        }
950        return python_sys_modules_read_identity(
951            parent.child_by_field_name("subscript"),
952            parent,
953            &proof,
954            budget,
955        );
956    }
957    if parent.kind() == "attribute"
958        && parent
959            .child_by_field_name("object")
960            .is_some_and(|object| object.id() == table.id())
961    {
962        let Some(call) = parent.parent().filter(|call| {
963            call.kind() == "call"
964                && call
965                    .child_by_field_name("function")
966                    .is_some_and(|function| function.id() == parent.id())
967        }) else {
968            return escaped();
969        };
970        let method = parent
971            .child_by_field_name("attribute")
972            .map(|attribute| node_text(attribute, source));
973        match method {
974            Some("get" | "pop") => {
975                let args = python_sys_modules_arguments(call);
976                return python_sys_modules_read_identity(
977                    args.first().copied(),
978                    call,
979                    &proof,
980                    budget,
981                );
982            }
983            Some("setdefault") => {
984                let args = python_sys_modules_arguments(call);
985                let default = args
986                    .get(1)
987                    .copied()
988                    .or_else(|| python_sys_modules_keyword_argument(call, source, "default"));
989                if let Some(default) = default
990                    && python_sys_modules_stored_value(
991                        default,
992                        &proof,
993                        budget,
994                        &mut HashSet::default(),
995                    ) != PythonSysModulesStoredValue::Unrelated
996                {
997                    return escaped();
998                }
999                return python_sys_modules_read_identity(
1000                    args.first().copied(),
1001                    call,
1002                    &proof,
1003                    budget,
1004                );
1005            }
1006            Some("update") => {
1007                let Some(arguments) = call.child_by_field_name("arguments") else {
1008                    return escaped();
1009                };
1010                let mut cursor = arguments.walk();
1011                for argument in arguments.named_children(&mut cursor) {
1012                    let identity = if argument.kind() == "keyword_argument" {
1013                        argument
1014                            .child_by_field_name("value")
1015                            .map(|value| {
1016                                python_sys_modules_stored_value(
1017                                    value,
1018                                    &proof,
1019                                    budget,
1020                                    &mut HashSet::default(),
1021                                )
1022                            })
1023                            .unwrap_or(PythonSysModulesStoredValue::Unknown)
1024                    } else if argument.kind() == "dictionary_splat" {
1025                        PythonSysModulesStoredValue::Unknown
1026                    } else {
1027                        python_sys_modules_update_values(
1028                            argument,
1029                            &proof,
1030                            budget,
1031                            &mut HashSet::default(),
1032                        )
1033                    };
1034                    if identity != PythonSysModulesStoredValue::Unrelated {
1035                        return escaped();
1036                    }
1037                }
1038                return PythonSysModulesLookup::Unrelated;
1039            }
1040            Some("clear" | "keys") => return PythonSysModulesLookup::Unrelated,
1041            _ => return escaped(),
1042        }
1043    }
1044    escaped()
1045}
1046
1047fn python_sys_modules_arguments(call: Node<'_>) -> Vec<Node<'_>> {
1048    let Some(arguments) = call.child_by_field_name("arguments") else {
1049        return Vec::new();
1050    };
1051    let mut cursor = arguments.walk();
1052    arguments
1053        .named_children(&mut cursor)
1054        .filter(|argument| argument.kind() != "keyword_argument")
1055        .collect()
1056}
1057
1058fn python_sys_modules_keyword_argument<'tree>(
1059    call: Node<'tree>,
1060    source: &str,
1061    name: &str,
1062) -> Option<Node<'tree>> {
1063    let arguments = call.child_by_field_name("arguments")?;
1064    let mut cursor = arguments.walk();
1065    arguments
1066        .named_children(&mut cursor)
1067        .find(|argument| {
1068            argument.kind() == "keyword_argument"
1069                && argument
1070                    .child_by_field_name("name")
1071                    .is_some_and(|key| node_text(key, source) == name)
1072        })
1073        .and_then(|argument| argument.child_by_field_name("value"))
1074}
1075
1076fn python_sys_modules_read_identity<'tree>(
1077    key: Option<Node<'tree>>,
1078    read: Node<'tree>,
1079    proof: &PythonSysModulesProofContext<'_, 'tree>,
1080    budget: &mut PythonRuntimeBudget,
1081) -> PythonSysModulesLookup<'tree> {
1082    let Some(key) = key else {
1083        return PythonSysModulesLookup::Escape(node_source_range(read));
1084    };
1085    if key.kind() == "identifier" && node_text(key, proof.source) == "__name__" {
1086        return if proof.source_module == proof.module {
1087            PythonSysModulesLookup::Target(read)
1088        } else {
1089            PythonSysModulesLookup::Unrelated
1090        };
1091    }
1092    let Some(key_value) = crate::decode_python_string_value(key, proof.source).ok() else {
1093        if python_sys_modules_guard_excludes_module(key, read, proof, budget) {
1094            return PythonSysModulesLookup::Unrelated;
1095        }
1096        return PythonSysModulesLookup::Target(read);
1097    };
1098    if key_value == proof.module {
1099        PythonSysModulesLookup::Target(read)
1100    } else {
1101        PythonSysModulesLookup::Unrelated
1102    }
1103}
1104
1105#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1106enum PythonGuardTruth {
1107    True,
1108    False,
1109    Unknown,
1110}
1111
1112/// Whether a structured dominating condition excludes this module name from
1113/// a dynamic `sys.modules` read. Unknown expressions retain the possible alias.
1114fn python_sys_modules_guard_excludes_module(
1115    key: Node<'_>,
1116    read: Node<'_>,
1117    proof: &PythonSysModulesProofContext<'_, '_>,
1118    budget: &mut PythonRuntimeBudget,
1119) -> bool {
1120    if key.kind() != "identifier" {
1121        return false;
1122    }
1123    let key_name = node_text(key, proof.source);
1124    let mut child = read;
1125    let mut ancestor = read.parent();
1126    while let Some(node) = ancestor {
1127        if !budget.visit() {
1128            return false;
1129        }
1130        if matches!(node.kind(), "if_statement" | "elif_clause")
1131            && let (Some(condition), Some(consequence)) = (
1132                node.child_by_field_name("condition"),
1133                node.child_by_field_name("consequence"),
1134            )
1135            && consequence.start_byte() <= read.start_byte()
1136            && read.end_byte() <= consequence.end_byte()
1137            && python_sys_modules_key_binding_unchanged(key, condition, read, proof, budget)
1138            && python_sys_modules_guard_truth(
1139                condition,
1140                key_name,
1141                proof.module,
1142                proof.source,
1143                budget,
1144            ) == PythonGuardTruth::False
1145        {
1146            return true;
1147        }
1148        if node.kind() == "block"
1149            && let Some(statement) = python_sys_modules_direct_child_containing(node, child)
1150            && python_sys_modules_prior_continue_guard_excludes(
1151                node, statement, key, read, key_name, proof, budget,
1152            )
1153        {
1154            return true;
1155        }
1156        child = node;
1157        ancestor = node.parent();
1158    }
1159    false
1160}
1161
1162/// Prove that the key has the same local binding in the guard and at the read.
1163/// The existing per-scope binding inventory used by module-alias propagation
1164/// counts every write in the lexical scope; this stricter source proof also
1165/// rejects any write or call on the path from the guard to the read.
1166fn python_sys_modules_key_binding_unchanged(
1167    key: Node<'_>,
1168    guard: Node<'_>,
1169    read: Node<'_>,
1170    proof: &PythonSysModulesProofContext<'_, '_>,
1171    budget: &mut PythonRuntimeBudget,
1172) -> bool {
1173    let name = node_text(key, proof.source);
1174    let guard_reference = python_sys_modules_guard_key_reference(guard, name, proof.source, budget);
1175    let Some(guard_reference) = guard_reference else {
1176        return false;
1177    };
1178    let scope = python_alias_scope(key);
1179    let key_resolution = python_name_resolution_at(name, key, proof.source);
1180    let guard_resolution = python_name_resolution_at(name, guard_reference, proof.source);
1181    let same_lexical_binding = key_resolution == guard_resolution
1182        && (key_resolution == PythonLexicalNameResolution::Local
1183            || (scope.kind() == "module"
1184                && key_resolution == PythonLexicalNameResolution::Unbound));
1185    if python_alias_scope(guard_reference).id() != scope.id()
1186        || python_alias_scope(read).id() != scope.id()
1187        || !same_lexical_binding
1188    {
1189        return false;
1190    }
1191    if !python_sys_modules_has_one_binding(proof, scope, name) {
1192        return false;
1193    }
1194    let start = guard.start_byte();
1195    let guard_end = guard.end_byte();
1196    let end = read.start_byte();
1197    let mut stack = vec![scope.child_by_field_name("body").unwrap_or(scope)];
1198    while let Some(node) = stack.pop() {
1199        if !budget.visit() {
1200            return false;
1201        }
1202        if node.start_byte() >= end || node.end_byte() <= start {
1203            continue;
1204        }
1205        if python_direct_scope_bindings_bounded(node, proof.source, || budget.visit()).is_none_or(
1206            |bindings| {
1207                bindings.into_iter().any(|binding| {
1208                    node_text(binding.declaration, proof.source) == name
1209                        && start <= binding.declaration.start_byte()
1210                        && binding.declaration.start_byte() < end
1211                })
1212            },
1213        ) {
1214            return false;
1215        }
1216        let mut cursor = node.walk();
1217        for child in node.named_children(&mut cursor) {
1218            if matches!(child.kind(), "call")
1219                && guard_end <= child.start_byte()
1220                && child.end_byte() < end
1221            {
1222                return false;
1223            }
1224        }
1225        stack.extend(python_executable_scope_children(node));
1226    }
1227    true
1228}
1229
1230fn python_sys_modules_guard_key_reference<'tree>(
1231    condition: Node<'tree>,
1232    name: &str,
1233    source: &str,
1234    budget: &mut PythonRuntimeBudget,
1235) -> Option<Node<'tree>> {
1236    let mut stack = vec![condition];
1237    while let Some(node) = stack.pop() {
1238        if !budget.visit() {
1239            return None;
1240        }
1241        if node.kind() == "identifier" && node_text(node, source) == name {
1242            return Some(node);
1243        }
1244        let mut cursor = node.walk();
1245        stack.extend(node.named_children(&mut cursor));
1246    }
1247    None
1248}
1249
1250fn python_sys_modules_direct_child_containing<'tree>(
1251    block: Node<'tree>,
1252    child: Node<'tree>,
1253) -> Option<Node<'tree>> {
1254    let mut direct = child;
1255    while direct
1256        .parent()
1257        .is_some_and(|parent| parent.id() != block.id())
1258    {
1259        direct = direct.parent()?;
1260    }
1261    direct
1262        .parent()
1263        .is_some_and(|parent| parent.id() == block.id())
1264        .then_some(direct)
1265}
1266
1267fn python_sys_modules_prior_continue_guard_excludes(
1268    block: Node<'_>,
1269    read_statement: Node<'_>,
1270    key: Node<'_>,
1271    read: Node<'_>,
1272    key_name: &str,
1273    proof: &PythonSysModulesProofContext<'_, '_>,
1274    budget: &mut PythonRuntimeBudget,
1275) -> bool {
1276    let mut cursor = block.walk();
1277    for statement in block.named_children(&mut cursor) {
1278        if statement.id() == read_statement.id() {
1279            break;
1280        }
1281        if statement.kind() != "if_statement" {
1282            continue;
1283        }
1284        let Some(condition) = statement.child_by_field_name("condition") else {
1285            continue;
1286        };
1287        if condition.kind() != "not_operator"
1288            || !python_sys_modules_guard_has_direct_abrupt_consequence(statement)
1289            || !python_sys_modules_key_binding_unchanged(key, condition, read, proof, budget)
1290        {
1291            continue;
1292        }
1293        let Some(guard_condition) = condition.named_child(0) else {
1294            continue;
1295        };
1296        if python_sys_modules_guard_truth(
1297            guard_condition,
1298            key_name,
1299            proof.module,
1300            proof.source,
1301            budget,
1302        ) == PythonGuardTruth::False
1303        {
1304            return true;
1305        }
1306    }
1307    false
1308}
1309
1310fn python_sys_modules_guard_has_direct_abrupt_consequence(statement: Node<'_>) -> bool {
1311    let Some(consequence) = statement.child_by_field_name("consequence") else {
1312        return false;
1313    };
1314    let mut cursor = consequence.walk();
1315    let mut statements = consequence.named_children(&mut cursor);
1316    let Some(only) = statements.next() else {
1317        return false;
1318    };
1319    statements.next().is_none()
1320        && matches!(
1321            only.kind(),
1322            "break_statement" | "continue_statement" | "return_statement" | "raise_statement"
1323        )
1324}
1325
1326fn python_sys_modules_guard_truth(
1327    condition: Node<'_>,
1328    key_name: &str,
1329    module: &str,
1330    source: &str,
1331    budget: &mut PythonRuntimeBudget,
1332) -> PythonGuardTruth {
1333    let mut pending = vec![(condition, false)];
1334    let mut results = HashMap::<usize, PythonGuardTruth>::default();
1335    while let Some((node, visited)) = pending.pop() {
1336        if !budget.visit() {
1337            return PythonGuardTruth::Unknown;
1338        }
1339        if !visited
1340            && matches!(
1341                node.kind(),
1342                "boolean_operator" | "not_operator" | "parenthesized_expression"
1343            )
1344        {
1345            pending.push((node, true));
1346            let mut cursor = node.walk();
1347            for child in node.named_children(&mut cursor) {
1348                pending.push((child, false));
1349            }
1350            continue;
1351        }
1352        let value = match node.kind() {
1353            "boolean_operator" => {
1354                let Some(left) = node.child_by_field_name("left") else {
1355                    return PythonGuardTruth::Unknown;
1356                };
1357                let Some(right) = node.child_by_field_name("right") else {
1358                    return PythonGuardTruth::Unknown;
1359                };
1360                let Some(operator) = node.child_by_field_name("operator") else {
1361                    return PythonGuardTruth::Unknown;
1362                };
1363                let (Some(left), Some(right)) = (results.get(&left.id()), results.get(&right.id()))
1364                else {
1365                    return PythonGuardTruth::Unknown;
1366                };
1367                match node_text(operator, source) {
1368                    "and" => python_guard_and(*left, *right),
1369                    "or" => python_guard_or(*left, *right),
1370                    _ => PythonGuardTruth::Unknown,
1371                }
1372            }
1373            "not_operator" => node
1374                .named_child(0)
1375                .and_then(|child| results.get(&child.id()).copied())
1376                .map(python_guard_not)
1377                .unwrap_or(PythonGuardTruth::Unknown),
1378            "parenthesized_expression" => node
1379                .named_child(0)
1380                .and_then(|child| results.get(&child.id()).copied())
1381                .unwrap_or(PythonGuardTruth::Unknown),
1382            _ => python_sys_modules_guard_atom(node, key_name, module, source),
1383        };
1384        results.insert(node.id(), value);
1385    }
1386    results
1387        .get(&condition.id())
1388        .copied()
1389        .unwrap_or(PythonGuardTruth::Unknown)
1390}
1391
1392fn python_guard_and(left: PythonGuardTruth, right: PythonGuardTruth) -> PythonGuardTruth {
1393    match (left, right) {
1394        (PythonGuardTruth::False, _) | (_, PythonGuardTruth::False) => PythonGuardTruth::False,
1395        (PythonGuardTruth::True, PythonGuardTruth::True) => PythonGuardTruth::True,
1396        _ => PythonGuardTruth::Unknown,
1397    }
1398}
1399
1400fn python_guard_or(left: PythonGuardTruth, right: PythonGuardTruth) -> PythonGuardTruth {
1401    match (left, right) {
1402        (PythonGuardTruth::True, _) | (_, PythonGuardTruth::True) => PythonGuardTruth::True,
1403        (PythonGuardTruth::False, PythonGuardTruth::False) => PythonGuardTruth::False,
1404        _ => PythonGuardTruth::Unknown,
1405    }
1406}
1407
1408fn python_guard_not(value: PythonGuardTruth) -> PythonGuardTruth {
1409    match value {
1410        PythonGuardTruth::True => PythonGuardTruth::False,
1411        PythonGuardTruth::False => PythonGuardTruth::True,
1412        PythonGuardTruth::Unknown => PythonGuardTruth::Unknown,
1413    }
1414}
1415
1416fn python_sys_modules_guard_atom(
1417    node: Node<'_>,
1418    key_name: &str,
1419    module: &str,
1420    source: &str,
1421) -> PythonGuardTruth {
1422    if node.kind() == "parenthesized_expression" {
1423        return PythonGuardTruth::Unknown;
1424    }
1425    if node.kind() == "comparison_operator" {
1426        return python_sys_modules_guard_comparison(node, key_name, module, source);
1427    }
1428    if node.kind() == "call" {
1429        return python_sys_modules_guard_string_method(node, key_name, module, source);
1430    }
1431    PythonGuardTruth::Unknown
1432}
1433
1434fn python_sys_modules_guard_comparison(
1435    node: Node<'_>,
1436    key_name: &str,
1437    module: &str,
1438    source: &str,
1439) -> PythonGuardTruth {
1440    let mut cursor = node.walk();
1441    let operands = node.named_children(&mut cursor).collect::<Vec<_>>();
1442    let mut cursor = node.walk();
1443    let operators = node
1444        .children_by_field_name("operators", &mut cursor)
1445        .collect::<Vec<_>>();
1446    let ([left, right], [operator]) = (operands.as_slice(), operators.as_slice()) else {
1447        return PythonGuardTruth::Unknown;
1448    };
1449    let operator = node_text(*operator, source);
1450    let key_left = left.kind() == "identifier" && node_text(*left, source) == key_name;
1451    let key_right = right.kind() == "identifier" && node_text(*right, source) == key_name;
1452    if key_left || key_right {
1453        let literal = if key_left {
1454            crate::decode_python_string_value(*right, source)
1455        } else {
1456            crate::decode_python_string_value(*left, source)
1457        };
1458        if matches!(operator, "==" | "!=")
1459            && let Ok(literal) = literal
1460        {
1461            let matches = literal == module;
1462            return python_guard_bool(if operator == "==" { matches } else { !matches });
1463        }
1464        if operator == "in" && key_left {
1465            return python_sys_modules_literal_membership(*right, module, source);
1466        }
1467        if operator == "in"
1468            && key_right
1469            && let Ok(literal) = literal
1470        {
1471            return python_guard_bool(module.contains(&literal));
1472        }
1473    }
1474    PythonGuardTruth::Unknown
1475}
1476
1477fn python_sys_modules_literal_membership(
1478    collection: Node<'_>,
1479    module: &str,
1480    source: &str,
1481) -> PythonGuardTruth {
1482    if !matches!(collection.kind(), "tuple" | "list" | "set") {
1483        return PythonGuardTruth::Unknown;
1484    }
1485    let mut cursor = collection.walk();
1486    let mut contains = false;
1487    for value in collection.named_children(&mut cursor) {
1488        let Ok(literal) = crate::decode_python_string_value(value, source) else {
1489            return PythonGuardTruth::Unknown;
1490        };
1491        contains |= literal == module;
1492    }
1493    python_guard_bool(contains)
1494}
1495
1496fn python_sys_modules_guard_string_method(
1497    call: Node<'_>,
1498    key_name: &str,
1499    module: &str,
1500    source: &str,
1501) -> PythonGuardTruth {
1502    let Some(function) = call.child_by_field_name("function") else {
1503        return PythonGuardTruth::Unknown;
1504    };
1505    if function.kind() != "attribute"
1506        || !function
1507            .child_by_field_name("object")
1508            .is_some_and(|object| {
1509                object.kind() == "identifier" && node_text(object, source) == key_name
1510            })
1511    {
1512        return PythonGuardTruth::Unknown;
1513    }
1514    let Some(method) = function
1515        .child_by_field_name("attribute")
1516        .map(|name| node_text(name, source))
1517    else {
1518        return PythonGuardTruth::Unknown;
1519    };
1520    let Some(arguments) = call.child_by_field_name("arguments") else {
1521        return PythonGuardTruth::Unknown;
1522    };
1523    let mut cursor = arguments.walk();
1524    let mut values = arguments.named_children(&mut cursor);
1525    let Some(first) = values.next() else {
1526        return PythonGuardTruth::Unknown;
1527    };
1528    if values.next().is_some() || first.kind() == "keyword_argument" {
1529        return PythonGuardTruth::Unknown;
1530    }
1531    let matches = if let Ok(literal) = crate::decode_python_string_value(first, source) {
1532        match method {
1533            "startswith" => module.starts_with(&literal),
1534            "endswith" => module.ends_with(&literal),
1535            _ => return PythonGuardTruth::Unknown,
1536        }
1537    } else if matches!(first.kind(), "tuple" | "list" | "set") {
1538        let mut cursor = first.walk();
1539        let mut matched = false;
1540        for value in first.named_children(&mut cursor) {
1541            let Ok(literal) = crate::decode_python_string_value(value, source) else {
1542                return PythonGuardTruth::Unknown;
1543            };
1544            matched |= match method {
1545                "startswith" => module.starts_with(&literal),
1546                "endswith" => module.ends_with(&literal),
1547                _ => return PythonGuardTruth::Unknown,
1548            };
1549        }
1550        matched
1551    } else {
1552        return PythonGuardTruth::Unknown;
1553    };
1554    python_guard_bool(matches)
1555}
1556
1557fn python_guard_bool(value: bool) -> PythonGuardTruth {
1558    if value {
1559        PythonGuardTruth::True
1560    } else {
1561        PythonGuardTruth::False
1562    }
1563}
1564
1565#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1566enum PythonSysModulesStoredValue {
1567    Target,
1568    Unrelated,
1569    Unknown,
1570}
1571
1572struct PythonSysModulesFacts<'tree> {
1573    aliases: HashSet<String>,
1574    assignments: HashMap<(usize, usize, String), Vec<Node<'tree>>>,
1575    binding_counts: HashMap<(usize, usize, String), usize>,
1576    classes: Vec<PythonExternalClassAlias>,
1577}
1578
1579struct PythonSysModulesProofContext<'a, 'tree> {
1580    source: &'a str,
1581    source_module: &'a str,
1582    module: &'a str,
1583    bindings: &'a [crate::imports::PythonImportBinding],
1584    facts: &'a PythonSysModulesFacts<'tree>,
1585}
1586
1587fn python_sys_modules_has_one_binding(
1588    proof: &PythonSysModulesProofContext<'_, '_>,
1589    scope: Node<'_>,
1590    name: &str,
1591) -> bool {
1592    let count = proof
1593        .facts
1594        .binding_counts
1595        .get(&(scope.start_byte(), scope.end_byte(), name.to_owned()))
1596        .copied()
1597        .unwrap_or_default()
1598        + proof
1599            .bindings
1600            .iter()
1601            .filter(|binding| {
1602                binding.local_name == name
1603                    && binding.scope_start_byte == scope.start_byte()
1604                    && binding.scope_end_byte == scope.end_byte()
1605            })
1606            .count();
1607    count == 1
1608}
1609
1610fn python_sys_modules_value_scope<'tree>(node: Node<'tree>, source: &str) -> Node<'tree> {
1611    let nearest = python_alias_scope(node);
1612    let name = python_static_path(node, source)
1613        .and_then(|(root, _)| (root.kind() == "identifier").then_some(node_text(root, source)))
1614        .unwrap_or_default();
1615    match python_name_resolution_at(name, node, source) {
1616        PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal => nearest,
1617        PythonLexicalNameResolution::Global => {
1618            let mut scope = nearest;
1619            while scope.kind() != "module" {
1620                scope = python_alias_scope(scope.parent().expect("lexical scope has a parent"));
1621            }
1622            scope
1623        }
1624        PythonLexicalNameResolution::Unbound if nearest.kind() == "class_definition" => nearest,
1625        PythonLexicalNameResolution::Unbound => {
1626            let mut scope = nearest;
1627            while scope.kind() != "module" {
1628                scope = python_alias_scope(scope.parent().expect("lexical scope has a parent"));
1629            }
1630            scope
1631        }
1632    }
1633}
1634
1635fn python_sys_modules_stored_value<'tree>(
1636    value: Node<'tree>,
1637    context: &PythonSysModulesProofContext<'_, 'tree>,
1638    budget: &mut PythonRuntimeBudget,
1639    seen: &mut HashSet<(usize, usize, String)>,
1640) -> PythonSysModulesStoredValue {
1641    let source = context.source;
1642    let module = context.module;
1643    let bindings = context.bindings;
1644    let facts = context.facts;
1645    if !budget.visit() {
1646        return PythonSysModulesStoredValue::Unknown;
1647    }
1648    if python_sys_modules_table(value, source).is_some() {
1649        return PythonSysModulesStoredValue::Unknown;
1650    }
1651    if let Some(table) = python_sys_modules_read_table(value, source) {
1652        if python_imported_path(table, source, bindings)
1653            .is_none_or(|path| path != ["sys", "modules"])
1654        {
1655            return PythonSysModulesStoredValue::Unknown;
1656        }
1657        return match python_sys_modules_read_identity(
1658            if value.kind() == "subscript" {
1659                value.child_by_field_name("subscript")
1660            } else {
1661                python_sys_modules_arguments(value).first().copied()
1662            },
1663            value,
1664            context,
1665            budget,
1666        ) {
1667            PythonSysModulesLookup::Target(_) => PythonSysModulesStoredValue::Target,
1668            PythonSysModulesLookup::Unrelated => PythonSysModulesStoredValue::Unrelated,
1669            PythonSysModulesLookup::Escape(_) => PythonSysModulesStoredValue::Unknown,
1670        };
1671    }
1672    if matches!(
1673        value.kind(),
1674        "string"
1675            | "concatenated_string"
1676            | "integer"
1677            | "float"
1678            | "true"
1679            | "false"
1680            | "none"
1681            | "list"
1682            | "tuple"
1683            | "set"
1684            | "dictionary"
1685            | "list_comprehension"
1686            | "set_comprehension"
1687            | "dictionary_comprehension"
1688            | "generator_expression"
1689    ) {
1690        return PythonSysModulesStoredValue::Unrelated;
1691    }
1692    if let Some(path) = python_imported_path(value, source, bindings) {
1693        let root_name = python_static_path(value, source)
1694            .map(|(root, _)| node_text(root, source))
1695            .unwrap_or_default();
1696        let scope = python_sys_modules_value_scope(value, source);
1697        if facts.binding_counts.contains_key(&(
1698            scope.start_byte(),
1699            scope.end_byte(),
1700            root_name.to_owned(),
1701        )) {
1702            return PythonSysModulesStoredValue::Unknown;
1703        }
1704        return if path.join(".") == module {
1705            PythonSysModulesStoredValue::Target
1706        } else {
1707            PythonSysModulesStoredValue::Unrelated
1708        };
1709    }
1710    if value.kind() == "call" {
1711        let Some(function) = value.child_by_field_name("function") else {
1712            return PythonSysModulesStoredValue::Unknown;
1713        };
1714        if python_imported_path(function, source, bindings)
1715            .is_some_and(|path| path == ["types", "ModuleType"])
1716        {
1717            return PythonSysModulesStoredValue::Unrelated;
1718        }
1719        if let Some((root, path)) = python_static_path(function, source)
1720            && path.len() == 1
1721            && python_class_alias_visible_at(node_text(root, source), root, source, &facts.classes)
1722        {
1723            return PythonSysModulesStoredValue::Unrelated;
1724        }
1725        return PythonSysModulesStoredValue::Unknown;
1726    }
1727    let Some((root, path)) = python_static_path(value, source) else {
1728        return PythonSysModulesStoredValue::Unknown;
1729    };
1730    if path.len() == 1 && facts.aliases.contains(node_text(root, source)) {
1731        return PythonSysModulesStoredValue::Target;
1732    }
1733    if path.len() == 1
1734        && python_class_alias_visible_at(node_text(root, source), root, source, &facts.classes)
1735    {
1736        return PythonSysModulesStoredValue::Unrelated;
1737    }
1738    let scope = python_sys_modules_value_scope(value, source);
1739    if scope.kind() == "function_definition"
1740        && scope
1741            .child_by_field_name("parameters")
1742            .is_some_and(|parameters| {
1743                let mut cursor = parameters.walk();
1744                parameters
1745                    .named_children(&mut cursor)
1746                    .filter_map(|parameter| python_parameter_name(parameter, source))
1747                    .any(|parameter| parameter == node_text(root, source))
1748            })
1749    {
1750        return PythonSysModulesStoredValue::Unknown;
1751    }
1752    if !python_sys_modules_has_one_binding(context, scope, node_text(root, source)) {
1753        return PythonSysModulesStoredValue::Unknown;
1754    }
1755    let key = (scope.start_byte(), scope.end_byte(), path.join("."));
1756    if !seen.insert(key.clone()) {
1757        return PythonSysModulesStoredValue::Unknown;
1758    }
1759    let Some([assignment]) = facts.assignments.get(&key).map(Vec::as_slice) else {
1760        return PythonSysModulesStoredValue::Unknown;
1761    };
1762    python_sys_modules_stored_value(*assignment, context, budget, seen)
1763}
1764
1765fn python_sys_modules_update_values<'tree>(
1766    mapping: Node<'tree>,
1767    context: &PythonSysModulesProofContext<'_, 'tree>,
1768    budget: &mut PythonRuntimeBudget,
1769    seen: &mut HashSet<(usize, usize, String)>,
1770) -> PythonSysModulesStoredValue {
1771    let source = context.source;
1772    let facts = context.facts;
1773    if mapping.kind() == "dictionary" {
1774        let mut cursor = mapping.walk();
1775        for pair in mapping.named_children(&mut cursor) {
1776            let Some(value) = pair.child_by_field_name("value") else {
1777                return PythonSysModulesStoredValue::Unknown;
1778            };
1779            let identity = python_sys_modules_stored_value(value, context, budget, seen);
1780            if identity != PythonSysModulesStoredValue::Unrelated {
1781                return identity;
1782            }
1783        }
1784        return PythonSysModulesStoredValue::Unrelated;
1785    }
1786    let Some((root, path)) = python_static_path(mapping, source) else {
1787        return PythonSysModulesStoredValue::Unknown;
1788    };
1789    let scope = python_sys_modules_value_scope(mapping, source);
1790    let key = (scope.start_byte(), scope.end_byte(), path.join("."));
1791    if !seen.insert(key.clone()) {
1792        return PythonSysModulesStoredValue::Unknown;
1793    }
1794    if !python_sys_modules_has_one_binding(context, scope, node_text(root, source)) {
1795        return PythonSysModulesStoredValue::Unknown;
1796    }
1797    let Some([assigned_value]) = facts.assignments.get(&key).map(Vec::as_slice) else {
1798        return PythonSysModulesStoredValue::Unknown;
1799    };
1800    python_sys_modules_update_values(*assigned_value, context, budget, seen)
1801}
1802
1803/// Collect aliases of the queried module reached through proven
1804/// `sys.modules` reads in this file. Only names and alias edges survive the
1805/// bounded, per-file walk.
1806fn python_sys_modules_aliases<'tree>(
1807    root: Node<'tree>,
1808    source: &str,
1809    source_module: &str,
1810    module: &str,
1811    bindings: &[crate::imports::PythonImportBinding],
1812    budget: &mut PythonRuntimeBudget,
1813) -> Option<PythonSysModulesFacts<'tree>> {
1814    let mut aliases_from = HashMap::<String, Vec<String>>::default();
1815    let mut assignments = HashMap::<(usize, usize, String), Vec<Node<'_>>>::default();
1816    let mut binding_counts = HashMap::<(usize, usize, String), usize>::default();
1817    let mut classes = Vec::new();
1818    let mut pending_table_aliases = Vec::<(Vec<String>, Option<Node<'_>>, Node<'_>)>::new();
1819    let mut stack = vec![root];
1820    while let Some(node) = stack.pop() {
1821        if !budget.visit() {
1822            return None;
1823        }
1824        if node.kind() == "class_definition"
1825            && let Some(name) = node.child_by_field_name("name")
1826        {
1827            let scope = python_alias_scope(node);
1828            classes.push(PythonExternalClassAlias {
1829                name: node_text(name, source).to_owned(),
1830                scope_start_byte: scope.start_byte(),
1831                scope_end_byte: scope.end_byte(),
1832            });
1833        }
1834        for binding in python_direct_scope_bindings_bounded(node, source, || budget.visit())? {
1835            if binding.kind != PythonDirectScopeBindingKind::Import {
1836                let scope = python_alias_scope(binding.declaration);
1837                let key = (
1838                    scope.start_byte(),
1839                    scope.end_byte(),
1840                    node_text(binding.declaration, source).to_owned(),
1841                );
1842                *binding_counts.entry(key).or_default() += 1;
1843            }
1844        }
1845        if matches!(node.kind(), "function_definition" | "lambda")
1846            && let Some(parameters) = node.child_by_field_name("parameters")
1847        {
1848            let scope = node;
1849            let mut cursor = parameters.walk();
1850            for parameter in parameters.named_children(&mut cursor) {
1851                if let Some(name) = python_parameter_name(parameter, source) {
1852                    *binding_counts
1853                        .entry((scope.start_byte(), scope.end_byte(), name))
1854                        .or_default() += 1;
1855                }
1856            }
1857        }
1858        if matches!(node.kind(), "assignment" | "annotated_assignment") {
1859            let (targets, value) = if node.kind() == "assignment" {
1860                crate::imports::python_assignment_chain(node)
1861            } else {
1862                (
1863                    node.child_by_field_name("left").into_iter().collect(),
1864                    node.child_by_field_name("right"),
1865                )
1866            };
1867            if let Some(value) = value {
1868                for target in targets {
1869                    if let Some((_, path)) = python_static_path(target, source) {
1870                        let scope = python_alias_scope(target);
1871                        let key = (scope.start_byte(), scope.end_byte(), path.join("."));
1872                        assignments.entry(key).or_default().push(value);
1873                    }
1874                }
1875            }
1876        }
1877        if node.kind() == "assignment" {
1878            let (targets, value) = crate::imports::python_assignment_chain(node);
1879            if let Some(value) = value
1880                && !targets.is_empty()
1881                && targets.iter().all(|target| target.kind() == "identifier")
1882            {
1883                let target_names = targets
1884                    .iter()
1885                    .map(|target| node_text(*target, source).to_owned())
1886                    .collect::<Vec<_>>();
1887                if python_sys_modules_read_table(value, source).is_some_and(|table| {
1888                    python_imported_path(table, source, bindings)
1889                        .is_some_and(|path| path == ["sys", "modules"])
1890                }) {
1891                    pending_table_aliases.push((
1892                        target_names,
1893                        if value.kind() == "subscript" {
1894                            value.child_by_field_name("subscript")
1895                        } else {
1896                            python_sys_modules_arguments(value).first().copied()
1897                        },
1898                        value,
1899                    ));
1900                } else if value.kind() == "identifier" {
1901                    aliases_from
1902                        .entry(node_text(value, source).to_owned())
1903                        .or_default()
1904                        .extend(target_names);
1905                }
1906            }
1907        }
1908        let mut cursor = node.walk();
1909        stack.extend(node.named_children(&mut cursor));
1910    }
1911
1912    let mut facts = PythonSysModulesFacts {
1913        aliases: HashSet::default(),
1914        assignments,
1915        binding_counts,
1916        classes,
1917    };
1918    let mut target_aliases = Vec::new();
1919    {
1920        let proof = PythonSysModulesProofContext {
1921            source,
1922            source_module,
1923            module,
1924            bindings,
1925            facts: &facts,
1926        };
1927        for (targets, key, read) in pending_table_aliases {
1928            if matches!(
1929                python_sys_modules_read_identity(key, read, &proof, budget),
1930                PythonSysModulesLookup::Target(_)
1931            ) {
1932                target_aliases.extend(targets);
1933            }
1934        }
1935    }
1936    facts.aliases.extend(target_aliases);
1937
1938    let mut pending = facts.aliases.iter().cloned().collect::<Vec<_>>();
1939    while let Some(source_name) = pending.pop() {
1940        if let Some(targets) = aliases_from.get(&source_name) {
1941            for target in targets {
1942                if facts.aliases.insert(target.clone()) {
1943                    pending.push(target.clone());
1944                }
1945            }
1946        }
1947    }
1948    Some(facts)
1949}
1950
1951fn python_sys_modules_read_use(
1952    read: Node<'_>,
1953    source: &str,
1954    member: &str,
1955    aliases: &HashSet<String>,
1956) -> Option<PythonModuleMemberMutation> {
1957    let mut expression = read;
1958    let mut direct_attribute = None;
1959    while let Some(parent) = expression.parent() {
1960        if parent.kind() != "attribute"
1961            || parent
1962                .child_by_field_name("object")
1963                .is_none_or(|object| object.id() != expression.id())
1964        {
1965            break;
1966        }
1967        if direct_attribute.is_none() {
1968            direct_attribute = Some(parent);
1969        }
1970        expression = parent;
1971    }
1972    if let Some(attribute) = direct_attribute {
1973        let name = attribute
1974            .child_by_field_name("attribute")
1975            .map(|name| node_text(name, source));
1976        if name == Some("__dict__") {
1977            return Some(PythonModuleMemberMutation::Escape(node_source_range(
1978                attribute,
1979            )));
1980        }
1981        if name == Some(member) && is_binding_target(attribute) {
1982            return Some(PythonModuleMemberMutation::Write(node_source_range(
1983                attribute,
1984            )));
1985        }
1986        return None;
1987    }
1988    if python_builtin_module_read(expression, source) {
1989        return None;
1990    }
1991    if read.parent().is_some_and(|parent| {
1992        parent.kind() == "assignment"
1993            && parent
1994                .child_by_field_name("right")
1995                .is_some_and(|right| right.id() == read.id())
1996            && crate::imports::python_assignment_chain(parent)
1997                .0
1998                .iter()
1999                .all(|target| {
2000                    target.kind() == "identifier" && aliases.contains(node_text(*target, source))
2001                })
2002    }) || read.parent().is_some_and(|parent| {
2003        parent.kind() == "expression_statement"
2004            && parent
2005                .named_child(0)
2006                .is_some_and(|expression| expression.id() == read.id())
2007    }) {
2008        return None;
2009    }
2010    Some(PythonModuleMemberMutation::Escape(node_source_range(read)))
2011}
2012
2013fn python_builtin_module_read(expression: Node<'_>, source: &str) -> bool {
2014    let Some(arguments) = expression
2015        .parent()
2016        .filter(|node| node.kind() == "argument_list")
2017    else {
2018        return false;
2019    };
2020    let Some(call) = arguments.parent().filter(|node| node.kind() == "call") else {
2021        return false;
2022    };
2023    let Some(function) = call.child_by_field_name("function") else {
2024        return false;
2025    };
2026    if function.kind() != "identifier"
2027        || !matches!(node_text(function, source), "getattr" | "hasattr")
2028        || matches!(
2029            python_name_resolution_at(node_text(function, source), function, source),
2030            PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2031        )
2032    {
2033        return false;
2034    }
2035    let mut cursor = arguments.walk();
2036    arguments
2037        .named_children(&mut cursor)
2038        .find(|argument| argument.kind() != "keyword_argument")
2039        .is_some_and(|first| first.id() == expression.id())
2040}
2041
2042fn python_module_table_alias_binding_use(
2043    node: Node<'_>,
2044    source: &str,
2045    aliases: &HashSet<String>,
2046) -> bool {
2047    if node.kind() != "identifier" {
2048        return false;
2049    }
2050    let name = node_text(node, source);
2051    if !aliases.contains(name) {
2052        return false;
2053    }
2054    if is_binding_target(node) {
2055        return true;
2056    }
2057    node.parent().is_some_and(|assignment| {
2058        assignment.kind() == "assignment"
2059            && assignment
2060                .child_by_field_name("right")
2061                .is_some_and(|right| right.id() == node.id())
2062            && crate::imports::python_assignment_chain(assignment)
2063                .0
2064                .iter()
2065                .all(|target| {
2066                    target.kind() == "identifier" && aliases.contains(node_text(*target, source))
2067                })
2068    })
2069}
2070
2071/// Find writes to one imported external class member in this file.
2072///
2073/// The class identities are qualified names taken from one exact active model
2074/// record (including its exact aliases). Import binders turn those identities
2075/// into local AST paths; source spelling alone never establishes the class.
2076/// Dynamic `setattr` names are treated as possible replacement writes.
2077pub fn python_external_class_member_mutation(
2078    root: Node<'_>,
2079    source: &str,
2080    class_identities: &[&str],
2081    member: &str,
2082    max_nodes: usize,
2083) -> Option<PythonModuleMemberMutation> {
2084    let bindings = python_import_bindings_from_tree(root, source);
2085    let Some(aliases) =
2086        python_external_class_aliases(root, source, &bindings, class_identities, max_nodes)
2087    else {
2088        return Some(PythonModuleMemberMutation::Budget);
2089    };
2090    let mut budget = PythonRuntimeBudget {
2091        remaining: max_nodes,
2092        exhausted: false,
2093        visited_nodes: 0,
2094    };
2095    let mut stack = vec![root];
2096    while let Some(node) = stack.pop() {
2097        if !budget.visit() {
2098            return Some(PythonModuleMemberMutation::Budget);
2099        }
2100        if node.kind() == "attribute"
2101            && node
2102                .child_by_field_name("attribute")
2103                .is_some_and(|attribute| node_text(attribute, source) == member)
2104            && is_binding_target(node)
2105            && let Some(object) = node.child_by_field_name("object")
2106            && python_external_class_expression(
2107                object,
2108                source,
2109                &bindings,
2110                class_identities,
2111                &aliases,
2112            )
2113        {
2114            return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2115        }
2116        if node.kind() == "call"
2117            && python_setattr_may_replace_class_member(
2118                node,
2119                source,
2120                &bindings,
2121                class_identities,
2122                member,
2123                &aliases,
2124            )
2125        {
2126            return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2127        }
2128        if node.kind() == "call"
2129            && python_mock_patch_may_replace_class_member(
2130                node,
2131                source,
2132                &bindings,
2133                class_identities,
2134                member,
2135                &aliases,
2136            )
2137        {
2138            return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2139        }
2140        if node.kind() == "identifier"
2141            && python_class_alias_visible_at(node_text(node, source), node, source, &aliases)
2142            && python_class_alias_escapes_at(node, source, &aliases)
2143        {
2144            return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2145        }
2146        let mut cursor = node.walk();
2147        stack.extend(node.named_children(&mut cursor));
2148    }
2149    None
2150}
2151
2152#[derive(Clone)]
2153struct PythonExternalClassAlias {
2154    name: String,
2155    scope_start_byte: usize,
2156    scope_end_byte: usize,
2157}
2158
2159fn python_external_class_aliases(
2160    root: Node<'_>,
2161    source: &str,
2162    bindings: &[crate::imports::PythonImportBinding],
2163    class_identities: &[&str],
2164    max_nodes: usize,
2165) -> Option<Vec<PythonExternalClassAlias>> {
2166    let mut budget = PythonRuntimeBudget {
2167        remaining: max_nodes,
2168        exhausted: false,
2169        visited_nodes: 0,
2170    };
2171    let mut assignments = Vec::new();
2172    let mut stack = vec![root];
2173    while let Some(node) = stack.pop() {
2174        if !budget.visit() {
2175            return None;
2176        }
2177        if matches!(node.kind(), "assignment" | "annotated_assignment")
2178            && let Some(target) = node.child_by_field_name("left")
2179            && target.kind() == "identifier"
2180            && let Some(value) = node.child_by_field_name("right")
2181        {
2182            let scope = python_alias_scope(node);
2183            assignments.push((
2184                node_text(target, source).to_owned(),
2185                value,
2186                scope.start_byte(),
2187                scope.end_byte(),
2188            ));
2189        }
2190        let mut cursor = node.walk();
2191        stack.extend(node.named_children(&mut cursor));
2192    }
2193
2194    let mut aliases = Vec::<PythonExternalClassAlias>::new();
2195    loop {
2196        let before = aliases.len();
2197        for (name, value, scope_start_byte, scope_end_byte) in &assignments {
2198            if python_external_class_expression(
2199                *value,
2200                source,
2201                bindings,
2202                class_identities,
2203                &aliases,
2204            ) && !aliases.iter().any(|alias| {
2205                alias.name == *name
2206                    && alias.scope_start_byte == *scope_start_byte
2207                    && alias.scope_end_byte == *scope_end_byte
2208            }) {
2209                aliases.push(PythonExternalClassAlias {
2210                    name: name.clone(),
2211                    scope_start_byte: *scope_start_byte,
2212                    scope_end_byte: *scope_end_byte,
2213                });
2214            }
2215        }
2216        if aliases.len() == before {
2217            return Some(aliases);
2218        }
2219    }
2220}
2221
2222fn python_alias_scope(mut node: Node<'_>) -> Node<'_> {
2223    while let Some(parent) = node.parent() {
2224        if matches!(
2225            parent.kind(),
2226            "module" | "class_definition" | "function_definition" | "lambda"
2227        ) {
2228            return parent;
2229        }
2230        node = parent;
2231    }
2232    node
2233}
2234
2235fn python_external_class_expression(
2236    expression: Node<'_>,
2237    source: &str,
2238    bindings: &[crate::imports::PythonImportBinding],
2239    class_identities: &[&str],
2240    aliases: &[PythonExternalClassAlias],
2241) -> bool {
2242    if python_imported_class_expression(expression, source, bindings, class_identities) {
2243        return true;
2244    }
2245    let Some((root, path)) = python_static_path(expression, source) else {
2246        return false;
2247    };
2248    path.len() == 1 && python_class_alias_visible_at(node_text(root, source), root, source, aliases)
2249}
2250
2251fn python_class_alias_visible_at(
2252    name: &str,
2253    reference: Node<'_>,
2254    source: &str,
2255    aliases: &[PythonExternalClassAlias],
2256) -> bool {
2257    if !aliases.iter().any(|alias| alias.name == name) {
2258        return false;
2259    }
2260    let mut current = reference;
2261    while let Some(scope) = nearest_scope(current) {
2262        if aliases.iter().any(|alias| {
2263            alias.name == name
2264                && alias.scope_start_byte == scope.start_byte()
2265                && alias.scope_end_byte == scope.end_byte()
2266        }) {
2267            return true;
2268        }
2269        match scope.kind() {
2270            "function_definition" | "lambda" => {
2271                let parameters = scope
2272                    .child_by_field_name("parameters")
2273                    .into_iter()
2274                    .flat_map(|parameters| {
2275                        let mut cursor = parameters.walk();
2276                        parameters
2277                            .named_children(&mut cursor)
2278                            .filter_map(|parameter| python_parameter_name(parameter, source))
2279                            .collect::<Vec<_>>()
2280                    });
2281                let Some(inventory) =
2282                    PythonLexicalScopeInventory::collect_bounded(scope, source, parameters, || {
2283                        true
2284                    })
2285                else {
2286                    return true;
2287                };
2288                match inventory.name_resolution_at(name, reference) {
2289                    PythonLexicalNameResolution::Local => return false,
2290                    PythonLexicalNameResolution::Global => break,
2291                    PythonLexicalNameResolution::Nonlocal
2292                    | PythonLexicalNameResolution::Unbound => {}
2293                }
2294            }
2295            "module" => return false,
2296            // Class names are not enclosing bindings for method bodies. A
2297            // class-body alias itself was handled above.
2298            "class_definition" => {}
2299            _ => unreachable!("nearest_scope returns only Python lexical scopes"),
2300        }
2301        current = scope;
2302    }
2303    aliases.iter().any(|alias| {
2304        alias.name == name
2305            && alias.scope_start_byte == 0
2306            && reference.start_byte() < alias.scope_end_byte
2307    })
2308}
2309
2310fn nearest_scope(mut node: Node<'_>) -> Option<Node<'_>> {
2311    while let Some(parent) = node.parent() {
2312        if matches!(
2313            parent.kind(),
2314            "module" | "class_definition" | "function_definition" | "lambda"
2315        ) {
2316            return Some(parent);
2317        }
2318        node = parent;
2319    }
2320    None
2321}
2322
2323fn python_class_alias_escapes_at(
2324    node: Node<'_>,
2325    source: &str,
2326    aliases: &[PythonExternalClassAlias],
2327) -> bool {
2328    let mut current = node;
2329    while let Some(parent) = current.parent() {
2330        if matches!(
2331            parent.kind(),
2332            "function_definition" | "lambda" | "class_definition" | "module"
2333        ) {
2334            return false;
2335        }
2336        if parent.kind() == "argument_list"
2337            && parent.parent().is_some_and(|call| call.kind() == "call")
2338        {
2339            return true;
2340        }
2341        if matches!(parent.kind(), "return_statement" | "yield") {
2342            return true;
2343        }
2344        if matches!(parent.kind(), "list" | "tuple" | "set" | "dictionary") {
2345            return true;
2346        }
2347        if matches!(parent.kind(), "assignment" | "annotated_assignment")
2348            && parent.child_by_field_name("right").is_some_and(|right| {
2349                right.start_byte() <= node.start_byte() && node.end_byte() <= right.end_byte()
2350            })
2351        {
2352            let Some(target) = parent.child_by_field_name("left") else {
2353                return true;
2354            };
2355            if target.kind() == "identifier"
2356                && python_class_alias_visible_at(node_text(target, source), target, source, aliases)
2357            {
2358                return false;
2359            }
2360            return true;
2361        }
2362        current = parent;
2363    }
2364    false
2365}
2366
2367fn python_setattr_may_replace_class_member(
2368    call: Node<'_>,
2369    source: &str,
2370    bindings: &[crate::imports::PythonImportBinding],
2371    class_identities: &[&str],
2372    member: &str,
2373    aliases: &[PythonExternalClassAlias],
2374) -> bool {
2375    let Some(function) = call.child_by_field_name("function") else {
2376        return false;
2377    };
2378    let imported_setattr = python_imported_path(function, source, bindings)
2379        .is_some_and(|path| path == ["builtins", "setattr"]);
2380    let builtin_setattr = python_static_path(function, source).is_some_and(|(root, path)| {
2381        path == ["setattr"]
2382            && !matches!(
2383                python_name_resolution_at(node_text(root, source), root, source),
2384                PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2385            )
2386    });
2387    if !imported_setattr && !builtin_setattr {
2388        return false;
2389    }
2390    let Some(arguments) = call.child_by_field_name("arguments") else {
2391        return false;
2392    };
2393    let mut cursor = arguments.walk();
2394    let positional = arguments
2395        .named_children(&mut cursor)
2396        .filter(|argument| argument.kind() != "keyword_argument")
2397        .collect::<Vec<_>>();
2398    let [object, name, ..] = positional.as_slice() else {
2399        return false;
2400    };
2401    if !python_external_class_expression(*object, source, bindings, class_identities, aliases) {
2402        return false;
2403    }
2404    python_plain_string_literal(*name, source).is_none_or(|name| name == member)
2405}
2406
2407fn python_mock_patch_may_replace_class_member(
2408    call: Node<'_>,
2409    source: &str,
2410    bindings: &[crate::imports::PythonImportBinding],
2411    class_identities: &[&str],
2412    member: &str,
2413    aliases: &[PythonExternalClassAlias],
2414) -> bool {
2415    let Some(function) = call.child_by_field_name("function") else {
2416        return false;
2417    };
2418    let Some(function_path) = python_imported_path(function, source, bindings) else {
2419        return false;
2420    };
2421    let patch_object = function_path == ["unittest", "mock", "patch", "object"];
2422    let patch_class_path = function_path == ["unittest", "mock", "patch"];
2423    if !patch_object && !patch_class_path {
2424        return false;
2425    }
2426    let Some(arguments) = call.child_by_field_name("arguments") else {
2427        return false;
2428    };
2429    let mut cursor = arguments.walk();
2430    let positional = arguments
2431        .named_children(&mut cursor)
2432        .filter(|argument| argument.kind() != "keyword_argument")
2433        .collect::<Vec<_>>();
2434    let Some(target) = positional.first().copied() else {
2435        return false;
2436    };
2437    if patch_object {
2438        let Some(name) = positional.get(1).copied() else {
2439            return false;
2440        };
2441        return python_external_class_expression(
2442            target,
2443            source,
2444            bindings,
2445            class_identities,
2446            aliases,
2447        ) && python_plain_string_literal(name, source).is_none_or(|name| name == member);
2448    }
2449    let Some(target_path) = python_plain_string_literal(target, source) else {
2450        return false;
2451    };
2452    class_identities
2453        .iter()
2454        .any(|class| target_path == *class || target_path == format!("{class}.{member}"))
2455}
2456
2457fn python_imported_path(
2458    expression: Node<'_>,
2459    source: &str,
2460    bindings: &[crate::imports::PythonImportBinding],
2461) -> Option<Vec<String>> {
2462    let (root, expression_path) = python_static_path(expression, source)?;
2463    let root_name = node_text(root, source);
2464    if matches!(
2465        python_name_resolution_at(root_name, root, source),
2466        PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2467    ) {
2468        return None;
2469    }
2470    let latest_import = bindings
2471        .iter()
2472        .filter(|binding| {
2473            binding.local_name == root_name
2474                && binding.start_byte <= root.start_byte()
2475                && binding.scope_start_byte <= root.start_byte()
2476                && root.end_byte() <= binding.scope_end_byte
2477        })
2478        .map(|binding| binding.start_byte)
2479        .max()?;
2480    let binding = bindings.iter().find(|binding| {
2481        binding.local_name == root_name
2482            && binding.start_byte == latest_import
2483            && binding.scope_start_byte <= root.start_byte()
2484            && root.end_byte() <= binding.scope_end_byte
2485    })?;
2486    let qualified_prefix = if binding.consumed_attributes > 0 {
2487        binding.qualified_name.split('.').next()?
2488    } else {
2489        binding.qualified_name.as_str()
2490    };
2491    let mut qualified_path = qualified_prefix
2492        .split('.')
2493        .map(str::to_owned)
2494        .collect::<Vec<_>>();
2495    qualified_path.extend(expression_path.into_iter().skip(1));
2496    Some(qualified_path)
2497}
2498
2499fn python_imported_class_expression(
2500    expression: Node<'_>,
2501    source: &str,
2502    bindings: &[crate::imports::PythonImportBinding],
2503    class_identities: &[&str],
2504) -> bool {
2505    let Some(path) = python_imported_path(expression, source, bindings) else {
2506        return false;
2507    };
2508    class_identities.iter().any(|identity| {
2509        identity
2510            .split('.')
2511            .map(str::to_owned)
2512            .eq(path.iter().cloned())
2513    })
2514}
2515
2516fn python_static_path<'tree>(
2517    node: Node<'tree>,
2518    source: &str,
2519) -> Option<(Node<'tree>, Vec<String>)> {
2520    let mut current = node;
2521    let mut attributes = Vec::new();
2522    loop {
2523        if current.kind() == "identifier" {
2524            attributes.reverse();
2525            let mut path = Vec::with_capacity(attributes.len() + 1);
2526            path.push(node_text(current, source).to_owned());
2527            path.extend(attributes);
2528            return Some((current, path));
2529        }
2530        if current.kind() != "attribute" {
2531            return None;
2532        }
2533        let attribute = current.child_by_field_name("attribute")?;
2534        attributes.push(node_text(attribute, source).to_owned());
2535        current = current.child_by_field_name("object")?;
2536    }
2537}
2538
2539/// Classify one occurrence of a name bound to the module object.
2540fn module_binding_use(
2541    node: Node<'_>,
2542    source: &str,
2543    bindings: &[crate::imports::PythonImportBinding],
2544    table_aliases: &HashSet<String>,
2545    module: &str,
2546    member: &str,
2547) -> Option<PythonModuleMemberMutation> {
2548    let Some(parent) = node.parent() else {
2549        return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2550    };
2551    if matches!(
2552        parent.kind(),
2553        "import_statement" | "import_from_statement" | "aliased_import" | "dotted_name"
2554    ) {
2555        return None;
2556    }
2557    let mut expression = node;
2558    while let Some(parent) = expression.parent() {
2559        if parent.kind() != "attribute"
2560            || parent
2561                .child_by_field_name("object")
2562                .is_none_or(|object| object.id() != expression.id())
2563        {
2564            break;
2565        }
2566        expression = parent;
2567    }
2568    if python_builtin_module_read(expression, source) {
2569        return None;
2570    }
2571    let imported_path = python_imported_path(expression, source, bindings);
2572    let table_alias_path = python_static_path(expression, source).and_then(|(root, suffix)| {
2573        let root_name = node_text(root, source);
2574        table_aliases.contains(root_name).then(|| {
2575            let mut resolved = vec![module.to_owned()];
2576            resolved.extend(suffix.into_iter().skip(1));
2577            resolved
2578        })
2579    });
2580    let Some(path) = imported_path.or(table_alias_path) else {
2581        return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2582    };
2583    let path = path.join(".");
2584    let target = format!("{module}.{member}");
2585    if path == target {
2586        return is_binding_target(expression)
2587            .then(|| PythonModuleMemberMutation::Write(node_source_range(expression)));
2588    }
2589    if path == format!("{module}.__dict__") || path == module {
2590        return Some(PythonModuleMemberMutation::Escape(node_source_range(
2591            expression,
2592        )));
2593    }
2594    if path.starts_with(&format!("{module}.")) {
2595        None
2596    } else {
2597        Some(PythonModuleMemberMutation::Escape(node_source_range(node)))
2598    }
2599}
2600
2601fn python_mock_patch_may_replace_module_member(
2602    call: Node<'_>,
2603    source: &str,
2604    bindings: &[crate::imports::PythonImportBinding],
2605    module: &str,
2606    member: &str,
2607) -> bool {
2608    let Some(function) = call.child_by_field_name("function") else {
2609        return false;
2610    };
2611    let Some(path) = python_imported_path(function, source, bindings) else {
2612        return false;
2613    };
2614    if path.len() < 2 || path[path.len() - 2] != "mock" || path[path.len() - 1] != "patch" {
2615        return false;
2616    }
2617    let Some(arguments) = call.child_by_field_name("arguments") else {
2618        return false;
2619    };
2620    let mut cursor = arguments.walk();
2621    let Some(target) = arguments.named_children(&mut cursor).next() else {
2622        return false;
2623    };
2624    let expected = format!("{module}.{member}");
2625    python_plain_string_literal(target, source) == Some(expected.as_str())
2626}
2627
2628#[cfg(test)]
2629mod tests {
2630    use super::*;
2631
2632    fn facts(source: &str) -> PythonRuntimeReadFacts {
2633        let mut parser = tree_sitter::Parser::new();
2634        parser
2635            .set_language(&tree_sitter_python::LANGUAGE.into())
2636            .expect("python grammar loads");
2637        let tree = parser.parse(source, None).expect("fixture parses");
2638        extract_python_runtime_reads(tree.root_node(), source, 64)
2639    }
2640
2641    /// The DataFlowBench template-1 positive shape.
2642    #[test]
2643    fn imported_module_environment_subscript_is_a_pristine_keyed_read() {
2644        let source = "import os\n\n\ndef run():\n    command = os.environ[\"DFB\"]\n    os.system(command)\n";
2645        let facts = facts(source);
2646        assert!(facts.complete, "{facts:#?}");
2647        let [read] = facts.reads.as_slice() else {
2648            panic!("exactly one keyed read: {facts:#?}");
2649        };
2650        assert_eq!(read.root_name, "os");
2651        assert_eq!(read.container, "environ");
2652        assert_eq!(
2653            read.access,
2654            PythonRuntimeAccessKey::Property("DFB".to_owned())
2655        );
2656        assert_eq!(
2657            read.lexical_resolution,
2658            PythonRuntimeRootResolution::ImportedModule
2659        );
2660        assert_eq!(read.mutation, PythonRuntimeMutationEvidence::NoKnownWrite);
2661        assert_eq!(
2662            read.accessor,
2663            PythonRuntimeAccessorCoverage::NoKnownAccessorEffects,
2664            "the only call runs after the read"
2665        );
2666        assert!(facts.writes.is_empty(), "{facts:#?}");
2667        // The structural seed is the container attribute, which is what an
2668        // RQL field-access query addresses.
2669        assert_eq!(
2670            &source[read.candidate_anchor.start_byte..read.candidate_anchor.end_byte],
2671            "os.environ"
2672        );
2673        assert_eq!(
2674            &source[read.range.start_byte..read.range.end_byte],
2675            "os.environ[\"DFB\"]"
2676        );
2677    }
2678
2679    #[test]
2680    fn imported_module_argv_index_is_a_keyed_read() {
2681        let facts = facts("import sys\n\n\ndef run():\n    return sys.argv[1]\n");
2682        let [read] = facts.reads.as_slice() else {
2683            panic!("exactly one keyed read: {facts:#?}");
2684        };
2685        assert_eq!(read.root_name, "sys");
2686        assert_eq!(read.container, "argv");
2687        assert_eq!(read.access, PythonRuntimeAccessKey::Index(1));
2688    }
2689
2690    /// A parameter named `os` is not the module, and the syntax proves it
2691    /// without comparing spellings anywhere else.
2692    #[test]
2693    fn a_shadowed_module_name_is_not_a_read() {
2694        let facts = facts("import os\n\n\ndef run(os):\n    return os.environ[\"DFB\"]\n");
2695        assert!(facts.reads.is_empty(), "{facts:#?}");
2696    }
2697
2698    #[test]
2699    fn a_local_container_name_is_not_a_read() {
2700        let facts =
2701            facts("def run():\n    environ = {\"DFB\": \"x\"}\n    return environ[\"DFB\"]\n");
2702        assert!(facts.reads.is_empty(), "{facts:#?}");
2703        assert!(facts.writes.is_empty(), "{facts:#?}");
2704    }
2705
2706    #[test]
2707    fn an_unrelated_module_spelled_os_is_not_a_read() {
2708        let facts = facts("from mypkg import os\n\n\ndef run():\n    return os.environ[\"DFB\"]\n");
2709        assert!(facts.reads.is_empty(), "{facts:#?}");
2710    }
2711
2712    #[test]
2713    fn a_dynamic_key_stays_a_read_with_no_static_key() {
2714        let facts = facts("import os\n\n\ndef run(name):\n    return os.environ[name]\n");
2715        let [read] = facts.reads.as_slice() else {
2716            panic!("one read: {facts:#?}");
2717        };
2718        assert_eq!(read.access, PythonRuntimeAccessKey::Dynamic);
2719    }
2720
2721    #[test]
2722    fn a_container_write_refutes_every_read_of_that_container() {
2723        let facts = facts(
2724            "import os\n\n\ndef run():\n    os.environ[\"A\"] = \"1\"\n    return os.environ[\"B\"]\n",
2725        );
2726        let [read] = facts.reads.as_slice() else {
2727            panic!("one read: {facts:#?}");
2728        };
2729        assert_eq!(read.mutation, PythonRuntimeMutationEvidence::KnownWrite);
2730        let [write] = facts.writes.as_slice() else {
2731            panic!("one write: {facts:#?}");
2732        };
2733        assert_eq!(write.target, PythonRuntimeWriteTarget::Key);
2734        assert_eq!(write.container.as_deref(), Some("environ"));
2735    }
2736
2737    #[test]
2738    fn a_container_method_receiver_refutes_the_pristine_claim() {
2739        let facts = facts(
2740            "import os\n\n\ndef run():\n    os.environ.update({})\n    return os.environ[\"B\"]\n",
2741        );
2742        let [read] = facts.reads.as_slice() else {
2743            panic!("one read: {facts:#?}");
2744        };
2745        assert_eq!(
2746            read.mutation,
2747            PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
2748        );
2749    }
2750
2751    #[test]
2752    fn a_module_object_escape_refutes_every_container() {
2753        let facts =
2754            facts("import os\n\n\ndef run(sink):\n    sink(os)\n    return os.environ[\"B\"]\n");
2755        let [read] = facts.reads.as_slice() else {
2756            panic!("one read: {facts:#?}");
2757        };
2758        assert_eq!(
2759            read.mutation,
2760            PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
2761        );
2762        assert!(
2763            facts
2764                .writes
2765                .iter()
2766                .any(|write| write.target == PythonRuntimeWriteTarget::ModuleEscape),
2767            "{facts:#?}"
2768        );
2769    }
2770
2771    #[test]
2772    fn a_call_before_the_read_leaves_accessor_coverage_open() {
2773        let facts = facts(
2774            "import os\n\n\ndef run(prepare):\n    prepare()\n    return os.environ[\"B\"]\n",
2775        );
2776        let [read] = facts.reads.as_slice() else {
2777            panic!("one read: {facts:#?}");
2778        };
2779        assert_eq!(
2780            read.accessor,
2781            PythonRuntimeAccessorCoverage::UnknownAccessorEffects
2782        );
2783    }
2784
2785    /// A call in a sibling function cannot run before this read, so it must
2786    /// not poison it: a reviewed sink call shares a module with its source.
2787    #[test]
2788    fn a_call_in_another_function_does_not_poison_the_read() {
2789        let source = "import os\n\n\ndef other(prepare):\n    prepare()\n\n\ndef run():\n    return os.environ[\"B\"]\n";
2790        let facts = facts(source);
2791        let [read] = facts.reads.as_slice() else {
2792            panic!("one read: {facts:#?}");
2793        };
2794        assert_eq!(
2795            read.accessor,
2796            PythonRuntimeAccessorCoverage::NoKnownAccessorEffects
2797        );
2798    }
2799
2800    #[test]
2801    fn a_call_later_in_the_same_loop_can_precede_the_read() {
2802        let source = "import os\n\n\ndef run(items, prepare):\n    for item in items:\n        value = os.environ[\"B\"]\n        prepare(value)\n";
2803        let facts = facts(source);
2804        let [read] = facts.reads.as_slice() else {
2805            panic!("one read: {facts:#?}");
2806        };
2807        assert_eq!(
2808            read.accessor,
2809            PythonRuntimeAccessorCoverage::UnknownAccessorEffects
2810        );
2811    }
2812
2813    #[test]
2814    fn an_aliased_import_keeps_the_absolute_module_identity() {
2815        let facts =
2816            facts("import os as operating\n\n\ndef run():\n    return operating.environ[\"B\"]\n");
2817        let [read] = facts.reads.as_slice() else {
2818            panic!("one read: {facts:#?}");
2819        };
2820        assert_eq!(read.binding_name, "operating");
2821        assert_eq!(read.root_name, "os");
2822    }
2823
2824    /// `import os` beside `import os.path` binds the same package, which the
2825    /// DataFlowBench propagator cell spells.
2826    #[test]
2827    fn a_submodule_import_agrees_with_its_package_binding() {
2828        let facts =
2829            facts("import os\nimport os.path\n\n\ndef run():\n    return os.environ[\"B\"]\n");
2830        let [read] = facts.reads.as_slice() else {
2831            panic!("one read: {facts:#?}");
2832        };
2833        assert_eq!(read.root_name, "os");
2834    }
2835
2836    #[test]
2837    fn a_module_scope_rebinding_of_the_import_name_is_not_a_read() {
2838        let facts = facts("import os\n\nos = None\n\n\ndef run():\n    return os.environ[\"B\"]\n");
2839        assert!(facts.reads.is_empty(), "{facts:#?}");
2840    }
2841}
2842
2843#[cfg(test)]
2844mod module_member_tests {
2845    use super::*;
2846
2847    fn mutation(source: &str, member: &str) -> Option<PythonModuleMemberMutation> {
2848        let mut parser = tree_sitter::Parser::new();
2849        parser
2850            .set_language(&tree_sitter_python::LANGUAGE.into())
2851            .expect("python grammar loads");
2852        let tree = parser.parse(source, None).expect("fixture parses");
2853        python_module_member_mutation(tree.root_node(), source, "test", "os", member, 4096)
2854    }
2855
2856    #[test]
2857    fn a_member_assignment_refuses_that_member() {
2858        let found = mutation(
2859            "import os\n\n\ndef patch(fake):\n    os.system = fake\n",
2860            "system",
2861        );
2862        assert!(
2863            matches!(found, Some(PythonModuleMemberMutation::Write(_))),
2864            "{found:?}"
2865        );
2866        // A write to another member leaves this one alone.
2867        assert_eq!(
2868            mutation(
2869                "import os\n\n\ndef patch(fake):\n    os.getcwd = fake\n",
2870                "system"
2871            ),
2872            None
2873        );
2874    }
2875
2876    #[test]
2877    fn a_module_object_escape_refuses_every_member() {
2878        for source in [
2879            "import os\n\n\ndef leak(sink):\n    sink(os)\n",
2880            "import os\n\n\nalias = os\n",
2881            "import os\nimport functools\n\n\ndef patch():\n    setattr(os, \"system\", None)\n",
2882        ] {
2883            let found = mutation(source, "system");
2884            assert!(
2885                matches!(found, Some(PythonModuleMemberMutation::Escape(_))),
2886                "{source}: {found:?}"
2887            );
2888        }
2889    }
2890
2891    #[test]
2892    fn a_matching_module_table_read_counts_its_attribute_write() {
2893        let found = mutation(
2894            "import sys\n\n\ndef patch(fake):\n    sys.modules[\"os\"].system = fake\n",
2895            "system",
2896        );
2897        assert!(
2898            matches!(found, Some(PythonModuleMemberMutation::Write(_))),
2899            "{found:?}"
2900        );
2901    }
2902
2903    #[test]
2904    fn a_file_that_never_binds_the_module_proves_the_absence() {
2905        assert_eq!(mutation("def run():\n    return 1\n", "system"), None);
2906    }
2907}
2908
2909#[cfg(test)]
2910mod shadowing_tests {
2911    use super::*;
2912
2913    fn facts(source: &str) -> PythonRuntimeReadFacts {
2914        let mut parser = tree_sitter::Parser::new();
2915        parser
2916            .set_language(&tree_sitter_python::LANGUAGE.into())
2917            .expect("python grammar loads");
2918        let tree = parser.parse(source, None).expect("fixture parses");
2919        extract_python_runtime_reads(tree.root_node(), source, 64)
2920    }
2921
2922    /// A parameter that spells the module name is not the module object, so it
2923    /// is not an escape and must not refute another function's pristine read.
2924    #[test]
2925    fn a_parameter_spelling_the_module_name_is_not_an_escape() {
2926        let source = "import os\n\n\ndef read_shadowed(os):\n    return os.environ[\"A\"]\n\n\ndef read_real():\n    return os.environ[\"B\"]\n";
2927        let facts = facts(source);
2928        assert!(facts.writes.is_empty(), "{facts:#?}");
2929        let [read] = facts.reads.as_slice() else {
2930            panic!("only the unshadowed read is a candidate: {facts:#?}");
2931        };
2932        assert_eq!(read.mutation, PythonRuntimeMutationEvidence::NoKnownWrite);
2933        assert_eq!(
2934            &source[read.key_range.start_byte..read.key_range.end_byte],
2935            "\"B\""
2936        );
2937    }
2938
2939    #[test]
2940    fn a_parameter_spelling_the_module_name_does_not_refuse_a_member() {
2941        let source =
2942            "import os\n\n\ndef helper(os):\n    return os\n\n\ndef run(c):\n    os.system(c)\n";
2943        let mut parser = tree_sitter::Parser::new();
2944        parser
2945            .set_language(&tree_sitter_python::LANGUAGE.into())
2946            .expect("python grammar loads");
2947        let tree = parser.parse(source, None).expect("fixture parses");
2948        assert_eq!(
2949            python_module_member_mutation(tree.root_node(), source, "test", "os", "system", 4096),
2950            None
2951        );
2952    }
2953}
2954
2955#[cfg(test)]
2956mod external_class_mutation_tests {
2957    use super::*;
2958    use brokk_bifrost_test_macros::hardwon;
2959
2960    fn mutation(source: &str) -> Option<PythonModuleMemberMutation> {
2961        let mut parser = tree_sitter::Parser::new();
2962        parser
2963            .set_language(&tree_sitter_python::LANGUAGE.into())
2964            .expect("python grammar loads");
2965        let tree = parser.parse(source, None).expect("fixture parses");
2966        python_external_class_member_mutation(
2967            tree.root_node(),
2968            source,
2969            &["paramiko.client.SSHClient"],
2970            "set_missing_host_key_policy",
2971            4096,
2972        )
2973    }
2974
2975    #[test]
2976    fn exact_imported_class_member_writes_are_found() {
2977        for source in [
2978            "from paramiko.client import SSHClient as Client\n\nClient.set_missing_host_key_policy = fake\n",
2979            "import paramiko.client as client\n\nclient.SSHClient.set_missing_host_key_policy = fake\n",
2980            "from paramiko.client import SSHClient as Client\n\nsetattr(Client, 'set_missing_host_key_policy', fake)\n",
2981            "from paramiko.client import SSHClient as Client\n\nsetattr(Client, name, fake)\n",
2982        ] {
2983            assert!(
2984                matches!(mutation(source), Some(PythonModuleMemberMutation::Write(_))),
2985                "{source}"
2986            );
2987        }
2988    }
2989
2990    #[test]
2991    #[hardwon("#4411: class aliases and mock patch APIs left external method dispatch exact")]
2992    fn class_aliases_and_mock_patch_apis_refute_method_closure() {
2993        for source in [
2994            "from paramiko.client import SSHClient as C\nAlias = C\nAlias.set_missing_host_key_policy = fake\n",
2995            "from paramiko.client import SSHClient as C\nAlias = C\nsetattr(Alias, 'set_missing_host_key_policy', fake)\n",
2996            "from unittest import mock\nfrom paramiko.client import SSHClient as C\nmock.patch.object(C, 'set_missing_host_key_policy', fake)\n",
2997            "from unittest.mock import patch\nfrom paramiko.client import SSHClient as C\npatch.object(C, 'set_missing_host_key_policy', fake)\n",
2998            "from unittest import mock\nmock.patch('paramiko.client.SSHClient.set_missing_host_key_policy', fake)\n",
2999        ] {
3000            assert!(
3001                matches!(mutation(source), Some(PythonModuleMemberMutation::Write(_))),
3002                "{source}"
3003            );
3004        }
3005    }
3006
3007    #[test]
3008    #[hardwon("#4411: an escaped class alias left external method dispatch exact")]
3009    fn an_escaped_class_alias_refutes_method_closure() {
3010        assert!(matches!(
3011            mutation("from paramiko.client import SSHClient as C\nAlias = C\nconsume(Alias)\n"),
3012            Some(PythonModuleMemberMutation::Escape(_))
3013        ));
3014    }
3015
3016    #[test]
3017    fn constructor_uses_and_same_named_workspace_classes_are_not_writes() {
3018        assert_eq!(
3019            mutation(
3020                "from paramiko.client import SSHClient as Client\n\n\ndef connect():\n    client = Client()\n    client.set_missing_host_key_policy(policy)\n"
3021            ),
3022            None
3023        );
3024        assert_eq!(
3025            mutation(
3026                "class SSHClient:\n    pass\n\nSSHClient.set_missing_host_key_policy = fake\n"
3027            ),
3028            None
3029        );
3030    }
3031}
3032
3033#[cfg(test)]
3034mod module_member_mutation_tests {
3035    use super::*;
3036    use brokk_bifrost_test_macros::hardwon;
3037
3038    fn mutation(source: &str) -> Option<PythonModuleMemberMutation> {
3039        mutation_in(source, "pkg.M")
3040    }
3041
3042    fn mutation_in(source: &str, source_module: &str) -> Option<PythonModuleMemberMutation> {
3043        mutation_for_module(source, source_module, "pkg.M")
3044    }
3045
3046    fn mutation_for_module(
3047        source: &str,
3048        source_module: &str,
3049        module: &str,
3050    ) -> Option<PythonModuleMemberMutation> {
3051        let mut parser = tree_sitter::Parser::new();
3052        parser
3053            .set_language(&tree_sitter_python::LANGUAGE.into())
3054            .expect("python grammar loads");
3055        let tree = parser.parse(source, None).expect("fixture parses");
3056        python_module_member_mutation(
3057            tree.root_node(),
3058            source,
3059            source_module,
3060            module,
3061            "NAME",
3062            4096,
3063        )
3064    }
3065
3066    #[test]
3067    #[hardwon(
3068        "#4622: writes through imported module aliases and mock patch APIs must keep a global class unknown"
3069    )]
3070    fn imported_module_member_mutation_covers_dynamic_write_forms() {
3071        for source in [
3072            "import pkg.M as target\ntarget.NAME = fake\n",
3073            "import pkg.M as target\ndel target.NAME\n",
3074            "import pkg.M as target\nsetattr(target, 'NAME', fake)\n",
3075            "import pkg.M as target\nsetattr(target, name, fake)\n",
3076            "import pkg.M as target\ntarget.__dict__['NAME'] = fake\n",
3077            "from unittest import mock\nmock.patch('pkg.M.NAME', fake)\n",
3078            "from unittest.mock import patch\npatch('pkg.M.NAME', fake)\n",
3079            "import pkg.M as target\nfrom unittest import mock\nmock.patch.object(target, 'NAME', fake)\n",
3080        ] {
3081            assert!(mutation(source).is_some(), "{source}");
3082        }
3083        assert_eq!(
3084            None,
3085            mutation("from unittest import mock\nmock.patch('pkg.Other.NAME', fake)\n")
3086        );
3087    }
3088
3089    #[test]
3090    #[hardwon("#4622: unrelated sys.modules reads blocked exact module-global class proofs")]
3091    fn sys_modules_reads_are_scoped_by_literal_module_identity() {
3092        assert_eq!(
3093            None,
3094            mutation("import sys\nother = sys.modules.get('torch')\n")
3095        );
3096        assert_eq!(None, mutation("import sys\nother = sys.modules['torch']\n"));
3097        assert!(matches!(
3098            mutation("import sys\nmodule = sys.modules.get('pkg.M')\nmodule.NAME = fake\n"),
3099            Some(PythonModuleMemberMutation::Write(_))
3100        ));
3101        assert!(matches!(
3102            mutation("import sys\nmodule = sys.modules['pkg.M']\nmodule.NAME = fake\n"),
3103            Some(PythonModuleMemberMutation::Write(_))
3104        ));
3105        assert_eq!(
3106            None,
3107            mutation("import sys\nmodule = sys.modules['pkg.M']\nconsume(module.NAME)\n")
3108        );
3109        assert!(matches!(
3110            mutation("import sys\nmodule = sys.modules[__name__]\nmodule.NAME = fake\n"),
3111            Some(PythonModuleMemberMutation::Write(_))
3112        ));
3113        assert_eq!(
3114            None,
3115            mutation_in("import sys\nmodule = sys.modules[__name__]\n", "pkg.Other")
3116        );
3117        for source in [
3118            "import sys\nsys.modules['pkg.M'] = fake\n",
3119            "import sys\nsys.modules.setdefault('pkg.M', fake)\n",
3120            "import sys\nsys.modules.update({'pkg.M': fake})\n",
3121            "import sys\nconsume(sys.modules)\n",
3122        ] {
3123            assert!(
3124                matches!(
3125                    mutation(source),
3126                    Some(PythonModuleMemberMutation::Escape(_))
3127                ),
3128                "{source}"
3129            );
3130        }
3131        for source in [
3132            "import sys\nmodule = sys.modules[module_name]\n",
3133            "import sys\ndel sys.modules['torch']\n",
3134            "import sys\nsys.modules.pop('torch')\n",
3135            "import sys\nsys.modules.clear()\n",
3136            "class Shim: pass\nimport sys\nsys.modules['shim'] = Shim()\n",
3137            "class Shim: pass\nimport sys\nsys.modules.update({'shim': Shim()})\n",
3138            "import sys\nimport types\nsys.modules['shim'] = types.ModuleType('shim')\n",
3139            "import sys\nsys.modules['literal'] = None\n",
3140            "import pkg.Other as shim\nimport sys\nsys.modules['shim'] = shim\n",
3141        ] {
3142            assert_eq!(mutation(source), None, "{source}");
3143        }
3144    }
3145
3146    #[test]
3147    #[hardwon("#4622: sys.modules values must be followed by module identity")]
3148    fn dynamic_module_table_reads_follow_only_writes_through_the_possible_alias() {
3149        assert!(matches!(
3150            mutation("import pkg.M as target\nimport sys\nsys.modules['shim'] = target\n"),
3151            Some(PythonModuleMemberMutation::Escape(_))
3152        ));
3153        assert_eq!(
3154            mutation("import sys\ntarget = sys.modules[module_name]\nconsume(target.NAME)\n"),
3155            None
3156        );
3157        assert_eq!(
3158            mutation(
3159                "import sys\ntarget = sys.modules[module_name]\nconsume(getattr(target, 'NAME'))\nconsume(hasattr(target, 'NAME'))\n"
3160            ),
3161            None
3162        );
3163        assert_eq!(
3164            mutation("import sys\nconsume(sys.modules[module_name].NAME)\n"),
3165            None
3166        );
3167        assert!(matches!(
3168            mutation("import sys\ntarget = sys.modules[module_name]\ntarget.NAME = fake\n"),
3169            Some(PythonModuleMemberMutation::Write(_))
3170        ));
3171    }
3172
3173    #[test]
3174    #[hardwon("#4622: a dominating string filter excludes unrelated dynamic sys.modules aliases")]
3175    fn dynamic_sys_modules_keys_respect_dominating_string_filters() {
3176        assert_eq!(
3177            mutation_for_module(
3178                "import sys\ndef patch():\n    for module_name in module_names:\n        if 'huggingface' in module_name or module_name.startswith('tqdm'):\n            target = sys.modules[module_name]\n            target.NAME = fake\n",
3179                "pkg.M",
3180                "backend.backends",
3181            ),
3182            None
3183        );
3184        assert!(matches!(
3185            mutation_for_module(
3186                "import sys\ndef patch():\n    for module_name in module_names:\n        if module_name.startswith('backend'):\n            target = sys.modules[module_name]\n            target.NAME = fake\n",
3187                "pkg.M",
3188                "backend.backends",
3189            ),
3190            Some(PythonModuleMemberMutation::Write(_))
3191        ));
3192        assert!(matches!(
3193            mutation_for_module(
3194                "import sys\ndef patch(allowed):\n    for module_name in module_names:\n        if module_name in allowed:\n            target = sys.modules[module_name]\n            target.NAME = fake\n",
3195                "pkg.M",
3196                "backend.backends",
3197            ),
3198            Some(PythonModuleMemberMutation::Write(_))
3199        ));
3200        assert!(matches!(
3201            mutation_for_module(
3202                "import sys\ndef patch():\n    for module_name in module_names:\n        if 'huggingface' in module_name:\n            module_name = 'backend.backends'\n            target = sys.modules[module_name]\n            target.NAME = fake\n",
3203                "pkg.M",
3204                "backend.backends",
3205            ),
3206            Some(PythonModuleMemberMutation::Write(_))
3207        ));
3208        assert_eq!(
3209            mutation_for_module(
3210                "import sys\ndef patch():\n    for module_name in module_names:\n        if not ('huggingface' in module_name):\n            continue\n        target = sys.modules[module_name]\n        target.NAME = fake\n",
3211                "pkg.M",
3212                "backend.backends",
3213            ),
3214            None
3215        );
3216    }
3217}