Add bounded source context analysis with lexical follow-up searches

config/project.example.toml

@@ -234,3 +234,14 @@ table = "creature_template_locale"
 id_column = "entry"
 name_column = "Name"
 columns = ["entry", "locale", "Name", "Title"]
+
+# Larger read-only source tool; registration naming is project-specific.
+[source_analysis]
+enabled = true
+context_lines = 8
+max_searches = 4
+max_matches = 12
+max_files = 8
+max_blocks = 12
+max_total_lines = 160
+registration_prefixes = ["AddSC_"]

docs/source-context-analysis-v1.md

@@ -0,0 +1,103 @@
+# Source-Kontextanalyse v1
+
+`project.query` kann mehrere zusammenhängende Suchen und Kontextabfragen bündeln:
+
+```json
+{"kind":"source.context","text":"23487","path_prefix":"src/server/"}
+```
+
+Das Werkzeug durchsucht ausschließlich Git-Objekte am vom Betreiber festgelegten
+Commit. Arbeitskopie, Branchwahl, Remote und Git-Optionen sind nicht durch die LLM
+änderbar. Ein vorhandener lokaler Checkout wird gelesen; dieses Werkzeug führt
+weder Fetch noch Build oder Sourceänderungen aus.
+
+## Ablauf und Ergebnis
+
+1. Ausgangsbegriff suchen; gültige Bezeichner werden als ganze Wörter gesucht,
+   andere Begriffe als feste, groß-/kleinschreibungssensitive Zeichenfolge.
+2. Begrenzt Kandidaten verfolgen: Bezeichner links einer einfachen Zuweisung des
+   Suchbegriffs sowie Funktionsnamen mit konfigurierten Registrierungsprefixen in
+   gefundenen Dateien. Das ist eine lexikalische Heuristik, keine Typauflösung.
+3. Kontextfenster um Treffer und Registrierungskandidaten lesen. Überlappende oder
+   angrenzende Fenster innerhalb derselben Datei zusammenführen.
+
+Dateipfad, Git-Blob, SHA-256 und Dateigröße werden pro Datei einmal angegeben.
+Die Ausschnitte bestehen aus `start_line` und einer Liste einfacher Codezeilen.
+Treffer enthalten Suchreferenzen und eine Angabe, ob ihr Kontext zurückgegeben
+wurde. Pro Suche bleiben Begriff, Bereich, Vollständigkeit und übersprungene
+Inhalte nachvollziehbar. Weitere Namen in zentralen Loader-Dateien werden nicht
+allein deshalb weiterverfolgt, weil dort die gesuchte Registrierung vorkommt.
+
+Das Ergebnis ist Kontext zu möglichen Definitionen, Verweisen und Registrierungen.
+Es behauptet keinen verifizierten Aufrufgraphen, keine tatsächliche Ausführung und
+keine Fehlerursache. Insbesondere können identische Zahlen andere Entitäten oder
+völlig andere Bedeutungen bezeichnen. Das zeigte auch der lokale Probeaufruf für
+23487: Die Fundstelle lag im Wintergrasp-Code; eine Verbindung zum Wolpertinger
+wird daraus nicht abgeleitet.
+
+## Konfiguration
+
+Das Werkzeug wird im Projektprofil ausdrücklich aktiviert:
+
+```toml
+[source_analysis]
+enabled = true
+context_lines = 8
+max_searches = 4
+max_matches = 12
+max_files = 8
+max_blocks = 12
+max_total_lines = 160
+registration_prefixes = ["AddSC_"]
+```
+
+`AddSC_` ist Adapterkonfiguration für Vcore, kein eingebautes Fachwissen im
+Kernsystem. Ohne solche Prefixe bleibt die allgemeine Such-/Aliasfunktion nutzbar.
+`extensions` kann die zugelassenen Textdateiendungen bestimmen; die Defaults
+enthalten unter anderem C/C++, SQL, Python, JavaScript/TypeScript, C#, Java, Go,
+Rust und Lua. Für größere Codebasen ist ein gezielter `path_prefix` sinnvoll.
+
+Die oben genannten Zahlen sind konfigurierbare Sicherheitsdefaults, keine
+Grenzen des Case-Paketformats. Der gemeinsame Zeitrahmen, Dateigrößen-, Suchbytes-
+und Ergebnisbytes-Grenzen kommen zusätzlich aus `ProjectLimits`. Interne Suchen
+erhalten ein begrenztes Teilbudget der Suchbytes. Höchstens die konfigurierte Zahl
+von Suchen, Dateien, Blöcken und Zeilen wird bearbeitet. Überlange Codezeilen
+werden gekürzt und mit ihrer ursprünglichen Zeilennummer als gekürzt markiert.
+Ein überschrittenes Ergebnisbytes-Limit oder ein Timeout erzeugt einen
+kontrollierten Fehler; es wird kein unmarkiert unvollständiger Capture gespeichert.
+
+## Vollständigkeit und Grenzen
+
+`limits_reached` nennt beispielsweise Treffer-, Folgeabfrage-, Datei-, Block- oder
+Zeilengrenzen. Solche Ergebnisse tragen `partial_configured_source_context`.
+`complete_configured_source_context` besagt nur, dass die gewählte begrenzte
+lexikalische Suchkette und ihre Kontextfenster ohne gemeldete Auslassungen
+verarbeitet wurden. Es bedeutet keine vollständige Implementierungsanalyse.
+
+Eine nicht gefundene Zahl schließt weder symbolische Konstanten mit anderer
+Schreibweise, Datenbanksteuerung, generierten Code, andere Dateitypen noch einen
+dynamischen Pfad aus. Eine Registrierungsfunktion ist ein Kandidat; die LLM muss
+ihre tatsächliche Rolle anhand des Codes bewerten. Bei unzureichenden Ausschnitten
+bleiben die bisherigen gezielten Source-Werkzeuge verfügbar.
+
+## Integration und Prüfung
+
+Neue Kontexte mit aktiviertem Werkzeug enthalten dessen Vertrag, Einstellungen
+und die Empfehlung, es vor getrennten Such-/Leseaufrufen zu verwenden. Eine Sperre
+der Grundwerkzeuge gibt es nicht. Bestehende Kontexte behalten ihren damaligen
+Vertrag. Der stabile Werkzeugkontext kann vom bestehenden Cache genutzt werden.
+
+Ein Aufruf ist eine budgetierte `project.query`-Aktion mit begrenzter interner
+Arbeit, ohne zusätzliche LLM-Zusammenfassung. Captures enthalten die ursprüngliche
+Anfrage, Commit-Bindung, Einstellungen samt Hash sowie sämtliche zurückgegebenen
+Codeausschnitte. Sie bleiben als unveränderliche Evidenz im Paket transportierbar.
+Abweichende Einstellungen gegenüber einem begonnenen Kontext werden abgelehnt.
+
+Lokale Tests prüfen Commitbindung trotz veränderter Arbeitskopie, Alias- und
+Registrierungskette, zusammengeführte Fenster, Python-Suche, leere Ergebnisse,
+Path Traversal, Argumentinjektion, Konfigurationsfehler, Zeit-/Bytegrenzen,
+gekennzeichnete Auslassungen, Budget und portable Wiederaufnahme. Ein simulierter
+LLM-Durchlauf prüft die Einbindung in die hybride Schleife. Die opt-in Prüfung am
+Vcore-Checkout vergleicht Hashes und Zeilen direkt mit den festgelegten Git-Objekten.
+Die Auswahl und Bewertung durch eine echte LLM bleibt Gegenstand des nächsten
+regulären Tests; eine Kostenersparnis wird damit noch nicht behauptet.

src/viano_bug_analysis/analysis/investigation_prompt.py

@@ -18,6 +18,9 @@ def output_schema(context=None, *, strict=False):
     if context and context['tool_schema'].get('x-project-bindings', {}).get('relation_profiles'):
         from viano_bug_analysis.tools.database_relations import extend_tool_schema
         schema['properties']['next'] = extend_tool_schema(schema['properties']['next'])
+    if context and context['tool_schema'].get('x-project-bindings', {}).get('source_analysis'):
+        from viano_bug_analysis.tools.source_context import extend_tool_schema
+        schema['properties']['next'] = extend_tool_schema(schema['properties']['next'])
     if context and context.get('prompt_version') in {'investigation_project_v12','investigation_project_v13','investigation_project_v14','investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}:
         schema['properties']['update']['properties']['causal_review'] = json.loads(
             files('viano_bug_analysis.schemas').joinpath('investigation-causal-review-v1.schema.json').read_text())

src/viano_bug_analysis/analysis/investigation_step.py

@@ -29,6 +29,9 @@ def _result_status(capture):
         status['completeness']='complete_for_configured_edges_only' if complete and result.get('coverage')=='complete_configured_relations' else 'partial_or_unknown'
         status['nodes']=[{k:node.get(k) for k in ('node','table','returned_rows','complete','upstream_complete','local_truncated')}
                          for node in items] if isinstance(items,list) else []
+    elif query.get('kind')=='source.context':
+        status['completeness']='complete_for_selected_lexical_context_only' if complete and result.get('coverage')=='complete_configured_source_context' else 'partial_or_unknown'
+        status['limits_reached']=result.get('limits_reached',[])
     # Unknown providers and selected source ranges never imply complete absence.
     return status
 

src/viano_bug_analysis/analysis/investigator.py

@@ -256,6 +256,11 @@ class Investigator:
                     'upstream truncation invalidates downstream absence claims.')
                 items = context['tool_schema']['properties']['requests']['items']
                 context['tool_schema']['properties']['requests']['items'] = extend_tool_schema(items)
+            if service.source_context:
+                from viano_bug_analysis.tools.source_context import extend_tool_schema
+                context['tool_schema']['x-project-bindings']['source_analysis'] = service.source_context.catalog()
+                items = context['tool_schema']['properties']['requests']['items']
+                context['tool_schema']['properties']['requests']['items'] = extend_tool_schema(items)
         else:
             variants=context["tool_schema"]["properties"]["requests"]["items"]["oneOf"]
             context["tool_schema"]["properties"]["requests"]["items"]["oneOf"]=[
@@ -583,6 +588,8 @@ class Investigator:
             raise ToolError("PROJECT_BINDING_CHANGED")
         if binding.get('relation_profiles') is not None and binding['relation_profiles'] != (service.relations.catalog() if service.relations else None):
             raise ToolError('PROJECT_BINDING_CHANGED')
+        if binding.get('source_analysis') is not None and binding['source_analysis'] != (service.source_context.catalog() if service.source_context else None):
+            raise ToolError('PROJECT_BINDING_CHANGED')
 
     def _invalid(self,state,budget,code,llm_ref,*,feedback=None):
         active=state["active"];h=state["hypotheses"][active["hypothesis_id"]]

src/viano_bug_analysis/schemas/investigation-request-v1.schema.json

@@ -709,6 +709,29 @@
                         "maxLength": 200
                       }
                     }
+                  },
+                  {
+                    "type": "object",
+                    "additionalProperties": false,
+                    "required": [
+                      "kind",
+                      "text",
+                      "path_prefix"
+                    ],
+                    "properties": {
+                      "kind": {
+                        "const": "source.context"
+                      },
+                      "text": {
+                        "type": "string",
+                        "minLength": 1,
+                        "maxLength": 200
+                      },
+                      "path_prefix": {
+                        "type": "string",
+                        "maxLength": 250
+                      }
+                    }
                   }
                 ]
               }

src/viano_bug_analysis/schemas/investigation-state-v1.schema.json

@@ -1710,6 +1710,29 @@
                                       "maxLength": 200
                                     }
                                   }
+                                },
+                                {
+                                  "type": "object",
+                                  "additionalProperties": false,
+                                  "required": [
+                                    "kind",
+                                    "text",
+                                    "path_prefix"
+                                  ],
+                                  "properties": {
+                                    "kind": {
+                                      "const": "source.context"
+                                    },
+                                    "text": {
+                                      "type": "string",
+                                      "minLength": 1,
+                                      "maxLength": 200
+                                    },
+                                    "path_prefix": {
+                                      "type": "string",
+                                      "maxLength": 250
+                                    }
+                                  }
                                 }
                               ]
                             }

src/viano_bug_analysis/schemas/project-capture-v1.schema.json

@@ -264,6 +264,29 @@
               "maxLength": 200
             }
           }
+        },
+        {
+          "type": "object",
+          "additionalProperties": false,
+          "required": [
+            "kind",
+            "text",
+            "path_prefix"
+          ],
+          "properties": {
+            "kind": {
+              "const": "source.context"
+            },
+            "text": {
+              "type": "string",
+              "minLength": 1,
+              "maxLength": 200
+            },
+            "path_prefix": {
+              "type": "string",
+              "maxLength": 250
+            }
+          }
         }
       ]
     },
@@ -298,7 +321,9 @@
             "partial_search",
             "complete_selected_text_search",
             "complete_configured_relations",
-            "partial_configured_relations"
+            "partial_configured_relations",
+            "complete_configured_source_context",
+            "partial_configured_source_context"
           ]
         },
         "truncated": {
@@ -348,6 +373,18 @@
             "null"
           ],
           "minimum": 1
+        },
+        "searches": {
+          "type": "array",
+          "items": {
+            "type": "object"
+          }
+        },
+        "limits_reached": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
         }
       }
     }

src/viano_bug_analysis/tools/project_query.py

@@ -35,7 +35,7 @@ class ProjectResearch:
     a URL, directory, credentials, database, branch, or arbitrary SQL.
     """
 
-    def __init__(self, project, *, source=None, database=None, limits=None, relations=None):
+    def __init__(self, project, *, source=None, database=None, limits=None, relations=None, source_analysis=None):
         self.project = copy.deepcopy(project)
         self.source, self.database = source, database
         self.limits = limits or ProjectLimits()
@@ -47,6 +47,10 @@ class ProjectResearch:
             self.relations = DatabaseRelations(database, relations) if relations else None
         except (ValueError, TypeError, KeyError):
             raise ToolError('PROJECT_RELATIONS_CONFIG') from None
+        from .source_context import SourceContextAnalysis
+        self.source_context = SourceContextAnalysis(source, source_analysis) if source_analysis is not None else None
+        if self.source_context and not self.source_context.config.enabled:
+            self.source_context = None
 
     @classmethod
     def load(cls, path):
@@ -55,7 +59,7 @@ class ProjectResearch:
         try:
             path = Path(path).resolve()
             config = tomllib.loads(path.read_text(encoding="utf-8"))
-            if set(config) - {"project", "source", "database", "limits", "enrichment", "entity_resolution", "relations", "relations_file"}:
+            if set(config) - {"project", "source", "database", "limits", "enrichment", "entity_resolution", "relations", "relations_file", "source_analysis"}:
                 raise ValueError()
             limits = ProjectLimits(**config.get("limits", {}))
             source = None
@@ -72,7 +76,8 @@ class ProjectResearch:
             relations = config.get('relations', {})
             if 'relations_file' in config:
                 relations = tomllib.loads((path.parent / config['relations_file']).read_text(encoding='utf-8'))['relations']
-            service = cls(config["project"], source=source, database=database, limits=limits, relations=relations)
+            service = cls(config["project"], source=source, database=database, limits=limits, relations=relations,
+                          source_analysis=config.get('source_analysis'))
             service.enrichment = copy.deepcopy(config.get("enrichment", {}))
             service.entity_resolution = copy.deepcopy(config.get("entity_resolution", {}))
             return service
@@ -89,7 +94,11 @@ class ProjectResearch:
 
     def query_schema(self):
         from .database_relations import extend_query_schema
-        return extend_query_schema(self.schema()) if self.relations else self.schema()
+        schema = extend_query_schema(self.schema()) if self.relations else self.schema()
+        if self.source_context:
+            from .source_context import extend_query_schema
+            schema = extend_query_schema(schema)
+        return schema
 
     def prepare(self, context, request):
         self.check_project(context.reader.manifest["project"])
@@ -111,6 +120,9 @@ class ProjectResearch:
         if arguments['kind'] == 'database.relations':
             self.relations.validate(arguments)
             return
+        if arguments['kind'] == 'source.context':
+            self.source_context.validate(arguments)
+            return
         provider.validate(arguments)
 
     def collect(self, arguments):
@@ -122,12 +134,20 @@ class ProjectResearch:
                        'consistency': 'single_repeatable_read_read_only_transaction',
                        'relation_plan': copy.deepcopy(self.relations.profiles[arguments['profile']]),
                        'relation_profile_sha256': self.relations.binding(arguments['profile'])}
+        if arguments['kind'] == 'source.context':
+            settings = self.source_context.catalog()
+            binding = {**binding, 'source_analysis': settings,
+                       'source_analysis_sha256': hashlib.sha256(json_bytes(settings)).hexdigest()}
         cache_key = hashlib.sha256(json_bytes({"arguments": arguments, "binding": binding})).hexdigest()
         # A request is frozen for the controller's validation/persistence cycle.
         # Durable resume uses the captured reply in the package, never this cache.
         if cache_key not in self._captures:
-            result = (self.relations.query(copy.deepcopy(arguments)) if arguments['kind'] == 'database.relations'
-                      else provider.query(copy.deepcopy(arguments)))
+            if arguments['kind'] == 'database.relations':
+                result = self.relations.query(copy.deepcopy(arguments))
+            elif arguments['kind'] == 'source.context':
+                result = self.source_context.query(copy.deepcopy(arguments))
+            else:
+                result = provider.query(copy.deepcopy(arguments))
             capture = {"format": "viano-project-capture", "format_version": 1,
                        "query": copy.deepcopy(arguments), "origin": binding, "result": result}
             from viano_bug_analysis.case.contracts import validate

src/viano_bug_analysis/tools/repository.py

@@ -112,7 +112,7 @@ class GitSourceProvider:
             if prefix != ".":
                 command.append(":(literal)" + prefix)
         else:
-            extensions = ["sql"] if args["kind"] == "source.schema" else ["cpp", "h", "hpp", "c", "sql", "txt", "md", "conf", "cmake"]
+            extensions = args.get('_extensions') or (["sql"] if args["kind"] == "source.schema" else ["cpp", "h", "hpp", "c", "sql", "txt", "md", "conf", "cmake"])
             command.extend(":(glob)**/*." + ext for ext in extensions)
         stream_queue = queue.Queue(maxsize=2)
         stop = threading.Event()
@@ -165,6 +165,8 @@ class GitSourceProvider:
                     continue
                 if not name.startswith(args["path_prefix"]) or (args["kind"] == "source.schema" and not name.endswith(".sql")):
                     continue
+                if '_extensions' in args and name.rsplit('.', 1)[-1] not in args['_extensions']:
+                    continue
                 if name not in sources:
                     try:
                         _, sources[name] = self._read(name)

src/viano_bug_analysis/tools/router.py

@@ -49,7 +49,7 @@ class InvestigationToolRouter:
             if branch['properties']['operation'].get('const') == 'project.query':
                 variants = branch['properties']['arguments']['oneOf']
                 branch['properties']['arguments']['oneOf'] = [v for v in variants
-                    if v['properties']['kind'].get('const') != 'database.relations']
+                    if v['properties']['kind'].get('const') not in {'database.relations', 'source.context'}]
         return schema
 
     def query(self, package, batch):

src/viano_bug_analysis/tools/source_context.py

@@ -0,0 +1,211 @@
+"""Bounded lexical context chains over immutable Git objects, never call graphs."""
+import copy
+import math
+import re
+import time
+from dataclasses import asdict, dataclass, replace
+
+from viano_bug_analysis.lifecycle.storage import json_bytes
+from .config import ToolError
+from .repository import safe_path
+
+
+def request_schema():
+    return {'type': 'object', 'additionalProperties': False,
+            'required': ['kind', 'text', 'path_prefix'], 'properties': {
+                'kind': {'const': 'source.context'},
+                'text': {'type': 'string', 'minLength': 1, 'maxLength': 200},
+                'path_prefix': {'type': 'string', 'maxLength': 250}}}
+
+
+def extend_query_schema(schema):
+    schema = copy.deepcopy(schema)
+    schema['oneOf'].append(request_schema())
+    return schema
+
+
+def extend_tool_schema(schema):
+    schema = copy.deepcopy(schema)
+    for branch in schema.get('oneOf', []):
+        if branch.get('properties', {}).get('operation', {}).get('const') == 'project.query':
+            branch['properties']['arguments'] = extend_query_schema(branch['properties']['arguments'])
+    return schema
+
+
+@dataclass(frozen=True)
+class SourceContextConfig:
+    enabled: bool = True
+    context_lines: int = 8
+    max_searches: int = 4
+    max_matches: int = 12
+    max_files: int = 8
+    max_blocks: int = 12
+    max_total_lines: int = 160
+    registration_prefixes: tuple = ()
+    extensions: tuple = ('cpp', 'h', 'hpp', 'c', 'sql', 'py', 'js', 'ts', 'tsx',
+                         'cs', 'go', 'rs', 'lua', 'java', 'cmake', 'txt')
+
+    def __post_init__(self):
+        bounds = {'context_lines': 100, 'max_searches': 12, 'max_matches': 100,
+                  'max_files': 30, 'max_blocks': 100, 'max_total_lines': 2000}
+        if type(self.enabled) is not bool:
+            raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG')
+        for key, maximum in bounds.items():
+            value = getattr(self, key)
+            if type(value) is not int or not 1 <= value <= maximum:
+                raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG')
+        for values, maximum, pattern in ((self.registration_prefixes, 8, r'[A-Za-z_][A-Za-z0-9_]{0,63}'),
+                                          (self.extensions, 30, r'[A-Za-z0-9]{1,12}')):
+            if (not isinstance(values, (list, tuple)) or len(values) > maximum
+                    or any(not isinstance(v, str) or not re.fullmatch(pattern, v) for v in values)):
+                raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG')
+        if not self.extensions:
+            raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG')
+
+
+class SourceContextAnalysis:
+    def __init__(self, provider, settings):
+        self.provider = provider
+        try:
+            self.config = SourceContextConfig(**settings)
+        except (TypeError, ValueError):
+            raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG') from None
+        if self.config.enabled and (not callable(getattr(provider, '_read', None)) or not callable(getattr(provider, 'query', None))):
+            raise ToolError('PROJECT_SOURCE_CONTEXT_CONFIG')
+
+    def catalog(self):
+        settings = {k: list(v) if isinstance(v, tuple) else v for k, v in asdict(self.config).items()}
+        return {'version': 1, 'settings': settings,
+                'preference': 'Prefer source.context for a symbol or entity identifier before separate '
+                              'search/read/registration calls. Returned links are lexical candidates, '
+                              'not verified calls, registration semantics or exclusions of dynamic paths.'}
+
+    def validate(self, args):
+        safe_path(args['path_prefix'], empty=True)
+        if not args['text'].strip() or any(ord(c) < 32 or ord(c) == 127 for c in args['text']):
+            raise ToolError('PROJECT_SOURCE_TEXT')
+
+    def query(self, args):
+        self.validate(args)
+        config, provider = self.config, self.provider
+        deadline = time.monotonic() + provider.limits.timeout_seconds
+        # Independent limits: do not mutate the shared provider during a query.
+        probe = copy.copy(provider)
+        files, anchors, searches, reached = {}, {}, [], set()
+        pending = [(args['text'], 'requested_term', None)]
+        seen, queued, registration_scanned = set(), {args['text']}, set()
+        read_bytes = 0
+
+        def remaining():
+            seconds = deadline - time.monotonic()
+            if seconds <= 0:
+                raise ToolError('PROJECT_SOURCE_TIMEOUT')
+            probe.limits = replace(provider.limits, timeout_seconds=max(1, math.ceil(seconds)),
+                                   max_search_bytes=max(1, provider.limits.max_search_bytes // config.max_searches))
+
+        def read_file(path):
+            nonlocal read_bytes
+            if path in files:
+                return files[path]
+            if len(files) >= config.max_files:
+                reached.add('file_limit'); return None
+            remaining()
+            text, origin = probe._read(path)
+            read_bytes += origin['size']
+            if read_bytes > provider.limits.max_search_bytes:
+                raise ToolError('PROJECT_SOURCE_INPUT_LIMIT')
+            files[path] = (text.splitlines(), origin)
+            return files[path]
+
+        def candidate(term, reason, parent):
+            if term in seen or term in queued: return
+            if len(queued) >= config.max_searches * config.max_matches:
+                reached.add('candidate_limit'); return
+            queued.add(term); pending.append((term, reason, parent))
+
+        while pending and len(searches) < config.max_searches:
+            term, reason, parent = pending.pop(0)
+            if term in seen:
+                continue
+            seen.add(term); remaining()
+            query = {'kind': 'source.symbol' if re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', term) else 'source.search',
+                     'text': term, 'path_prefix': args['path_prefix'], 'case_sensitive': True,
+                     'offset': 0, 'limit': min(config.max_matches, provider.limits.max_rows),
+                     '_extensions': list(config.extensions)}
+            result = probe.query(query)
+            qid = 'S' + str(len(searches) + 1)
+            search = {'id': qid, 'reason': reason, 'parent_search': parent, 'query': query,
+                      'coverage': result['coverage'], 'truncated': result['truncated'],
+                      'skipped_files': result.get('skipped_files', []),
+                      'returned_matches': len(result['items'])}
+            searches.append(search)
+            if result['truncated']: reached.add('match_limit')
+            if result.get('skipped_files') or result['coverage'] != 'complete_selected_text_search':
+                reached.add('incomplete_search')
+            for hit in result['items']:
+                path, line = hit['path'], hit['line']
+                content = read_file(path)
+                if content is None: continue
+                lines, origin = content
+                if not 1 <= line <= len(lines):
+                    raise ToolError('PROJECT_SOURCE_RANGE')
+                anchors.setdefault(path, {}).setdefault(line, set()).add(qid)
+                # Equality syntax is a candidate alias, not a semantic identity proof.
+                match = re.search(r'\b([A-Za-z_][A-Za-z0-9_]*)\s*=\s*' + re.escape(term) + r'(?![A-Za-z0-9_])', lines[line-1])
+                if match and match[1] != term:
+                    candidate(match[1], 'lexical_assignment_candidate', qid)
+                # Trusted project-specific prefixes select nearby registration context.
+                if (reason == 'lexical_registration_candidate' or path in registration_scanned
+                        or any(term.startswith(prefix) for prefix in config.registration_prefixes)): continue
+                registration_scanned.add(path)
+                for number, text in enumerate(lines, 1):
+                    if number % 256 == 0 and time.monotonic() >= deadline:
+                        raise ToolError('PROJECT_SOURCE_TIMEOUT')
+                    if len(anchors.get(path, {})) >= config.max_searches * config.max_matches:
+                        reached.add('registration_anchor_limit'); break
+                    for prefix in config.registration_prefixes:
+                        for found in re.finditer(r'\b(' + re.escape(prefix) + r'[A-Za-z0-9_]*)\s*\(', text):
+                            symbol = found[1]
+                            anchors.setdefault(path, {}).setdefault(number, set()).update({qid, 'registration_candidate'})
+                            if symbol != term: candidate(symbol, 'lexical_registration_candidate', qid)
+        if any(term not in seen for term, _, _ in pending): reached.add('followup_search_limit')
+
+        items, used_lines, used_blocks = [], 0, 0
+        for path, (lines, origin) in files.items():
+            ranges = []
+            for line in sorted(anchors.get(path, {})):
+                start, end = max(1, line-config.context_lines), min(len(lines), line+config.context_lines)
+                if ranges and start <= ranges[-1][1] + 1:
+                    ranges[-1][1] = max(ranges[-1][1], end)
+                else: ranges.append([start, end])
+            blocks = []
+            for start, end in ranges:
+                if used_blocks >= config.max_blocks: reached.add('block_limit'); break
+                count = min(end-start+1, config.max_total_lines-used_lines)
+                if count <= 0: reached.add('line_limit'); break
+                if count < end-start+1: reached.add('line_limit')
+                selected = lines[start-1:start-1+count]
+                truncated = [start+i for i, text in enumerate(selected) if len(text) > 2048]
+                if truncated: reached.add('line_text_limit')
+                blocks.append({'start_line': start, 'lines': [text[:2048] for text in selected],
+                               'truncated_lines': truncated})
+                used_lines += count; used_blocks += 1
+            visible = {start+i for b in blocks for start in [b['start_line']] for i in range(len(b['lines']))}
+            selected_anchors = [{'line': line, 'search_ids': sorted(ids-{'registration_candidate'}),
+                                 'registration_candidate': 'registration_candidate' in ids, 'context_returned': line in visible}
+                                for line, ids in sorted(anchors.get(path, {}).items())]
+            omitted = sum(not a['context_returned'] for a in selected_anchors)
+            if omitted: reached.add('omitted_anchors')
+            if blocks:
+                items.append({**origin, 'total_lines': len(lines), 'blocks': blocks,
+                              'anchors': selected_anchors, 'omitted_anchors': omitted})
+        remaining()
+        result = {'items': items, 'searches': searches, 'limits_reached': sorted(reached),
+                  'coverage': 'partial_configured_source_context' if reached else 'complete_configured_source_context',
+                  'truncated': bool(reached),
+                  'interpretation': 'Selected lexical searches and bounded context windows at the configured commit only. '
+                                    'Assignments and registration names are candidates, not verified semantic relationships. '
+                                    'No complete call graph, runtime observation, root cause or global absence claim.'}
+        if len(json_bytes(result)) > provider.limits.max_result_bytes:
+            raise ToolError('PROJECT_RESULT_LIMIT')
+        return result

tests/test_source_context.py

@@ -0,0 +1,204 @@
+import copy
+import json
+from pathlib import Path
+import subprocess
+
+import pytest
+from jsonschema import Draft202012Validator
+
+from viano_bug_analysis.tools import ToolError
+from viano_bug_analysis.tools.project_query import ProjectResearch, ProjectLimits
+from viano_bug_analysis.tools.repository import GitSourceProvider
+from test_case_builder import case_setup
+from test_investigation_tools import tool_case
+
+
+@pytest.fixture
+def source(tmp_path):
+    root=tmp_path/'source';(root/'src').mkdir(parents=True)
+    (root/'src/events.cpp').write_text('enum Ids {\nNPC_WILD = 23487,\n};\nvoid spawn() {\nsummon(NPC_WILD);\n}\nvoid AddSC_events() {\nregister_handler("npc_wild",spawn);\n}\n',encoding='utf-8')
+    (root/'src/loader.cpp').write_text('void AddSC_events();\nvoid load() { AddSC_events(); }\nvoid AddSC_unrelated();\n',encoding='utf-8')
+    (root/'src/unrelated.cpp').write_text('int NPC_WILD_LONGER;\n',encoding='utf-8')
+    (root/'src/module.py').write_text('def activate():\n    return "python context"\n\nactivate()\n',encoding='utf-8')
+    (root/'src/long.cpp').write_text('LONG_TARGET '+('x'*3000)+'\n',encoding='utf-8')
+    (root/'src/gaps.cpp').write_text('GAPPED\n'+('\n'*25)+'GAPPED\n',encoding='utf-8')
+    def git(*args):return subprocess.run(['git','-C',str(root),*args],check=True,capture_output=True).stdout.decode().strip()
+    git('init');git('add','.')
+    git('-c','user.name=Test','-c','user.email=test@example.invalid','commit','-m','fixture')
+    commit=git('rev-parse','HEAD')
+    (root/'src/events.cpp').write_text('mutable worktree content must be ignored\n',encoding='utf-8')
+    return GitSourceProvider(root,commit,'https://example.invalid/repo')
+
+
+def service(source,**settings):
+    return ProjectResearch({},source=source,source_analysis={'registration_prefixes':['AddSC_'],**settings})
+
+
+def query(text='23487',prefix='src/'):
+    return {'kind':'source.context','text':text,'path_prefix':prefix}
+
+
+def test_chain_alias_registration_context_is_compact_and_pinned(source):
+    s=service(source);capture=json.loads(s.collect(query()));result=capture['result']
+    assert [x['query']['text'] for x in result['searches']]==['23487','NPC_WILD','AddSC_events']
+    assert not result['truncated']
+    assert {i['path'] for i in result['items']}=={'src/events.cpp','src/loader.cpp'}
+    assert all(len(i['blocks'])==1 for i in result['items'])
+    events=result['items'][0]
+    assert events['blocks'][0]['start_line']==1
+    assert 'NPC_WILD = 23487,' in events['blocks'][0]['lines']
+    assert 'mutable worktree' not in json.dumps(result)
+    assert capture['origin']['commit']==source.commit
+    assert 'not verified semantic' in result['interpretation']
+    assert 'AddSC_unrelated' not in [s['query']['text'] for s in result['searches']]
+    assert s.collect(query())==s.collect(query())
+
+
+def test_python_and_empty_scoped_search(source):
+    s=service(source)
+    r=json.loads(s.collect(query('activate','')))['result']
+    assert [i['path'] for i in r['items']]==['src/module.py']
+    r=json.loads(s.collect(query('absent','src/module.py')))['result']
+    assert r['items']==[] and not r['truncated']
+    assert 'global absence' in r['interpretation']
+
+
+def test_registration_name_does_not_expand_unrelated_loader_symbols(source):
+    r=json.loads(service(source).collect(query('AddSC_events')))['result']
+    assert [q['query']['text'] for q in r['searches']]==['AddSC_events']
+    assert not r['truncated']
+
+
+def test_long_lines_and_missing_blocks_are_explicit(source):
+    r=json.loads(service(source).collect(query('LONG_TARGET')))['result']
+    assert 'line_text_limit' in r['limits_reached']
+    assert r['items'][0]['blocks'][0]['truncated_lines']==[1]
+    assert len(r['items'][0]['blocks'][0]['lines'][0])==2048
+    r=json.loads(service(source,context_lines=1,max_blocks=1).collect(query('GAPPED')))['result']
+    assert {'block_limit','omitted_anchors'} <= set(r['limits_reached'])
+    assert any(not a['context_returned'] for a in r['items'][0]['anchors'])
+
+
+@pytest.mark.parametrize('prefix',['../outside','/etc','C:/outside','.git/','a\\b','src/\x00'])
+def test_unsafe_paths_rejected_before_git(source,monkeypatch,prefix):
+    s=service(source)
+    monkeypatch.setattr(source,'query',lambda args:pytest.fail('must not access repository'))
+    with pytest.raises(ToolError):s.collect(query(prefix=prefix))
+
+
+@pytest.mark.parametrize('text',['','   ','x\ny','x\x00y','x\ty'])
+def test_invalid_text_is_controlled(source,text):
+    with pytest.raises(ToolError):service(source).collect(query(text))
+
+
+def test_no_regex_shell_or_git_options(source):
+    s=service(source)
+    assert json.loads(s.collect(query('.*')))['result']['items']==[]
+    assert json.loads(s.collect(query('--help')))['result']['items']==[]
+    with pytest.raises(ToolError):s.collect({**query(),'command':'git fetch'})
+    with pytest.raises(ToolError):s.collect({**query(),'_extensions':['cpp']})
+
+
+@pytest.mark.parametrize('settings',[{'max_searches':0},{'max_matches':True},{'max_files':1.0},
+    {'registration_prefixes':['.*']},{'extensions':[]},{'extensions':['cpp;bad']},{'unknown':1}])
+def test_invalid_configuration(source,settings):
+    with pytest.raises(ToolError):service(source,**settings)
+
+
+@pytest.mark.parametrize('settings,reason',[({'max_searches':1},'followup_search_limit'),
+    ({'max_matches':1},'match_limit'),({'max_files':1},'file_limit'),
+    ({'max_total_lines':1},'line_limit')])
+def test_limits_are_reported_without_claiming_complete(source,settings,reason):
+    r=json.loads(service(source,**settings).collect(query()))['result']
+    assert r['truncated'] and reason in r['limits_reached']
+    assert r['coverage']=='partial_configured_source_context'
+
+
+def test_nondefault_settings_binding_changes_and_disabled_tool_stays_hidden(source):
+    one=service(source);two=service(source,context_lines=2)
+    assert json.loads(one.collect(query()))['origin']['source_analysis_sha256']!=json.loads(two.collect(query()))['origin']['source_analysis_sha256']
+    disabled=ProjectResearch({},source=source,source_analysis={'enabled':False})
+    assert disabled.query_schema()==ProjectResearch.schema()
+    with pytest.raises(ToolError):disabled.collect(query())
+
+
+def test_byte_and_time_limits_are_controlled(source,monkeypatch):
+    source.limits=ProjectLimits(max_result_bytes=100)
+    s=service(source)
+    with pytest.raises(ToolError,match='RESULT_LIMIT'):s.collect(query())
+    assert not s._captures
+    import viano_bug_analysis.tools.source_context as module
+    clock=iter([0,100])
+    monkeypatch.setattr(module.time,'monotonic',lambda:next(clock))
+    with pytest.raises(ToolError,match='TIMEOUT'):s.collect(query())
+
+
+def test_source_context_strict_contract_and_old_replay(source):
+    from test_investigation_lean import context
+    from viano_bug_analysis.analysis.investigation_prompt import output_schema
+    c=context();c.pop('lean_session',None)
+    action={'kind':'tool','operation':'project.query','arguments':query()}
+    assert not Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
+    c['tool_schema']['x-project-bindings']['source_analysis']=service(source).source_context.catalog()
+    assert Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
+
+
+def test_changed_source_analysis_cannot_execute_a_frozen_context(source):
+    from types import SimpleNamespace
+    from viano_bug_analysis.analysis import Investigator
+    s=service(source)
+    c={'tool_schema':{'x-project-bindings':{'source':source.binding(),'database':None,
+        'source_analysis':s.source_context.catalog()}}}
+    loop=Investigator.__new__(Investigator);loop.router=SimpleNamespace(project_research=s)
+    loop._check_project_binding(c)
+    s.source_context=service(source,context_lines=1).source_context
+    with pytest.raises(ToolError,match='BINDING_CHANGED'):loop._check_project_binding(c)
+
+
+def test_budgeted_evidence_and_portable_hybrid_resume(source,tool_case,tmp_path):
+    from test_investigation_hybrid import HybridAdapter
+    from test_case_interpreter import gateway
+    from test_investigator import run,state
+    from viano_bug_analysis.analysis import Investigator,InvestigationConfig
+    from viano_bug_analysis.tools import InvestigationToolRouter
+    from viano_bug_analysis.package import PackageReader,PackageValidator
+    from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
+    from viano_bug_analysis.budget import BudgetManager
+    s=service(source)
+    with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project']
+    adapter=HybridAdapter()
+    adapter.bootstrap.change=lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':query()})
+    router=InvestigationToolRouter(project_research=s)
+    loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),router=router)
+    first=run(loop,steps=3)
+    assert all(h['iterations']==1 for h in state(first)['hypotheses'].values())
+    assert 'source_analysis' in adapter.calls[0]['input']['tool_schema']['project']
+    portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
+    final=run(Investigator(portable,gateway(adapter),router=router))
+    assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
+    from viano_bug_analysis.llm.cache_measurement import measure_package
+    measurement=measure_package(final.package)
+    assert measurement['unmatched_requests']==0 and measurement['all_exactly_reconstructed']
+    budget=BudgetManager(portable).snapshot('case-A','stable-package').state
+    assert all(a['status']=='settled' for h in budget['accounts'].values() for a in h['actions'].values())
+    with PackageReader(final.package) as reader:
+        paths=[p for p in reader.files if p.startswith('analysis/evidence/captures/')]
+        capture=next(json.loads(reader.read_bytes(p)) for p in paths if json.loads(reader.read_bytes(p))['query']['kind']=='source.context')
+        assert capture['result']['items'][0]['git_blob']
+        assert capture['origin']['commit']==source.commit
+
+
+def test_real_pinned_source_context():
+    import os
+    config=os.environ.get('VBA_TEST_PROJECT_CONFIG')
+    if not config:pytest.skip('Opt-in pinned local source checkout')
+    s=ProjectResearch.load(config)
+    capture=json.loads(s.collect(query('enum UnitFlags','src/server/')))
+    assert capture['origin']['commit']==s.source.commit
+    assert capture['result']['items']
+    for item in capture['result']['items']:
+        text,metadata=s.source._read(item['path'])
+        assert item['git_blob']==metadata['git_blob'] and item['sha256']==metadata['sha256']
+        lines=text.splitlines()
+        for block in item['blocks']:
+            assert block['lines']==[line[:2048] for line in lines[block['start_line']-1:block['start_line']-1+len(block['lines'])]]