Add bounded read-only database relationship analysis profiles

config/project-relations.vcore.toml

@@ -0,0 +1,34 @@
+# Trusted Vcore adapter data. Joins describe equality, not root-cause rules.
+[relations.creature_spawn_context]
+description = "Given creature_template.entry, inspect its template, bounded world spawns, event GUID links and event descriptions. Distribution covers returned spawn rows only. No dynamic script or live-state analysis."
+
+[[relations.creature_spawn_context.nodes]]
+name = "template"
+table = "creature_template"
+match_column = "entry"
+columns = ["entry", "name", "AIName", "ScriptName"]
+
+[[relations.creature_spawn_context.nodes]]
+name = "spawns"
+parent = "template"
+parent_column = "entry"
+table = "creature"
+match_column = "id"
+columns = ["guid", "id", "map", "zoneId", "areaId", "phaseMask"]
+group_by = ["map", "zoneId", "areaId", "phaseMask"]
+
+[[relations.creature_spawn_context.nodes]]
+name = "event_links"
+parent = "spawns"
+parent_column = "guid"
+table = "game_event_creature"
+match_column = "guid"
+columns = ["guid", "eventEntry"]
+
+[[relations.creature_spawn_context.nodes]]
+name = "events"
+parent = "event_links"
+parent_column = "eventEntry"
+table = "game_event"
+match_column = "eventEntry"
+columns = ["eventEntry", "description"]

config/project.example.toml

@@ -1,4 +1,5 @@
 # Operator-controlled bindings. The model cannot override these settings.
+relations_file = "project-relations.vcore.toml"
 [project]
 type = "wow"
 profile = "vcore"

docs/database-relations-v1.md

@@ -0,0 +1,108 @@
+# DB-Beziehungsanalyse v1
+
+Die Operation `project.query` akzeptiert bei konfigurierten Analyseprofilen:
+
+```json
+{"kind":"database.relations","profile":"creature_spawn_context","value":23487}
+```
+
+Die LLM bestimmt nur Profil und Startwert. Tabellen, Spalten, Gleichheitsbeziehungen
+und Projektionen kommen aus der vertrauenswürdigen Projektkonfiguration. Es gibt
+keine freie SQL-Eingabe, keine automatische Fremdschlüsselvermutung und keine
+eingebaute Diagnose. Grundwerkzeuge bleiben für weitere Fragen verfügbar.
+
+## Konfiguration und Systemgrenze
+
+`relations_file = "project-relations.vcore.toml"` steht auf der obersten Ebene
+der Projekt-TOML. Alternativ können Profile direkt unter `[relations]` definiert
+werden; beide Varianten zugleich sind unzulässig. Das Vcore-Beispiel ist getrennte
+Adapterkonfiguration. Der Kern kennt keine Kreaturen, Quests oder Gebietsnummern.
+
+Ein Profil hat eine Beschreibung und eine geordnete Liste benannter Knoten:
+
+- Der erste Knoten filtert `table.match_column` auf den Startwert.
+- Folgeknoten filtern ihre `match_column` auf die eindeutigen, nichtleeren Werte
+  von `parent_column` in den zurückgegebenen Zeilen eines früheren Knotens.
+- `columns` bestimmt die zurückzugebenden Spalten. `group_by` kann eine Verteilung
+  über höchstens fünf dieser Spalten ergänzen.
+
+Verzweigungen sind möglich, Zyklen nicht. Unterstützt sind direkte
+Gleichheitsbeziehungen; zusammengesetzte Schlüssel, rekursive Traversierung und
+Transformationen wie der Betrag negativer Event-IDs werden nicht automatisch
+interpretiert. Zusätzliche Profile können unabhängig von der Kernlogik ergänzt
+werden. Eine ausbleibende Zuordnung ist noch keine Feststellung eines DB-Fehlers.
+
+Das Vcore-Profil untersucht `creature_template` → `creature` →
+`game_event_creature` → `game_event`. Es prüft weder dynamische Spawnpfade im
+Sourcecode noch den aktuellen Zustand eines laufenden Servers. Koordinaten werden
+in diesem Profil nicht übertragen; sie sind bei Bedarf mit Grundwerkzeugen lesbar.
+
+## Grenzen, Vollständigkeit und Provenienz
+
+Pro Profil sind höchstens zwölf Knoten und dreißig projizierte Spalten je Knoten
+erlaubt. `ProjectLimits.max_rows` begrenzt jede Ergebnismenge, `timeout_seconds`
+die gesamte Analyse und einzelne SQL-Statements, `max_result_bytes` das Ergebnis.
+Dies sind Sicherheitsvorgaben, keine Grenzen des Case-Paketformats. Ein Profilaufruf
+ist eine budgetierte `project.query`-Aktion mit begrenzter interner Arbeit, keine
+Folge zusätzlicher LLM-Aufrufe. Die vorhandene Aktionskostenkonfiguration bleibt
+maßgeblich; dieses Werkzeug ändert keine Budgetwerte.
+
+Alle Abfragen eines Aufrufs erfolgen in einer gemeinsamen Read-only-Transaktion
+mit Repeatable Read und konsistentem Snapshot. Nur InnoDB/XtraDB-Tabellen sind
+dafür zugelassen; andere Engines erzeugen kontrolliert
+`PROJECT_RELATIONS_SNAPSHOT_UNSUPPORTED`. Die Transaktion wird zurückgerollt und
+geschlossen. Angefragte Identifikatoren und Spalten werden geprüft, Werte gebunden.
+Ganzzahlige Startspalten verlangen einen echten Integer, damit MySQL keine
+beliebigen Texte implizit zu IDs konvertiert.
+
+Jeder Knoten liefert genaue Filter, projizierte Zeilen, benötigte Spaltenmetadaten,
+Anzahl der zurückgegebenen Zeilen und getrennte Angaben zu lokaler Begrenzung und
+vollständiger Elternmenge. Verteilungen beziehen sich ausdrücklich nur auf die
+zurückgegebenen Zeilen. Eine begrenzte Elternmenge macht alle abhängigen Ergebnisse
+unvollständig, auch wenn deren eigene Seite nicht begrenzt wurde. Die tatsächliche
+Gesamtzahl außerhalb einer begrenzten Seite wird nicht behauptet.
+
+Ohne Elternschlüssel wird eine Folgeabfrage übersprungen und als solche markiert.
+Ein fehlender Startdatensatz schließt insbesondere verwaiste Zeilen in anderen
+Tabellen nicht aus. `complete_configured_relations` bedeutet ausschließlich:
+Die konkret konfigurierten Verbindungen wurden ohne Seitenbegrenzung verarbeitet.
+Es bedeutet weder vollständige Datenbankanalyse noch Ausschluss anderer Ursachen.
+Für größere Mengen stehen die bisherigen paginierbaren Grundwerkzeuge bereit.
+
+Der Capture enthält Snapshot-Bindung, vollständigen Profilplan und dessen Hash.
+Er wird wie bisher unveränderlich als Evidenz gespeichert und ist ohne erneute
+DB-Abfrage transportierbar. Änderungen am Profil ändern die Capture-Bindung;
+Änderungen gegenüber einem bereits begonnenen Untersuchungskontext werden
+abgelehnt. Abfragen werden während des Controller-Zyklus eingefroren. Die konkrete
+DB-Momentaufnahme hängt weiterhin von der ehrlich konfigurierten Snapshot-ID ab.
+
+## LLM-Anbindung und Kompatibilität
+
+Neue Kontexte mit Profilen erhalten Katalog, Argumentvertrag und die Empfehlung,
+passende größere Werkzeuge zuerst zu verwenden. Sie müssen diese nicht verwenden.
+Der bestehende Cache kann den Werkzeugkatalog wie andere stabile Kontextbestandteile
+wiederverwenden. Eine zusätzliche Zusammenfassungs-LLM gibt es nicht.
+
+Ohne Profile bleibt die bisherige Promptstruktur bestehen. Gespeicherte Kontexte
+behalten ihren damaligen Vertrag. Die portablen Request-, State- und Capture-Schemas
+akzeptieren die zusätzliche Operation. Die Prüfung numerischer Codingangaben
+berücksichtigt auch die hier gewonnenen Zeilen.
+
+## Lokale Prüfung
+
+Tests decken parametrisierte Abfragen, Allowlist, Konfigurationsfehler, begrenzte
+Elternmengen, leere Ergebnisse, Zeit-/Bytegrenzen, Snapshot-Engine-Prüfung,
+Paketvalidierung, Budgetierung und Wiederaufnahme ab. Ein simulierter LLM-Durchlauf
+prüft die Werkzeugauswahl in der hybriden Schleife.
+
+Der opt-in Test an der lokalen Vcore-Testdatenbank prüft für Entry 23487 eine
+Vorlage, achtzehn Spawns ausschließlich auf Map 1, achtzehn dazugehörige
+Event-Verknüpfungen und Event 24. Der projizierte Capture umfasst beim geprüften
+Backup ungefähr 8,3 KB. Das ist ein Funktionsnachweis, keine Kostenprognose und
+noch kein Nachweis der Werkzeugwahl durch eine echte LLM. Es wurden dafür keine
+kostenpflichtigen API-Aufrufe ausgeführt.
+
+```powershell
+$env:VBA_TEST_PROJECT_CONFIG = "config/project.local.toml"
+.\.venv\Scripts\python.exe -m pytest tests/test_database_relations.py -q
+```

src/viano_bug_analysis/analysis/database_value_grounding.py

@@ -19,7 +19,11 @@ def unsupported_assignments(payload,context):
             query=capture.get('query');result=capture.get('result')
             if not isinstance(query,dict) or not isinstance(result,dict):continue
             kind=query.get('kind')
-            if kind=='database.schema':
+            if kind=='database.relations':
+                for node in result.get('items',[]):
+                    for row in node.get('rows',[]):
+                        for column,value in row.items():remember(column,value)
+            if kind in {'database.schema','database.relations'}:
                 for item in capture.get('result',{}).get('items',[]):
                     if not isinstance(item,dict):continue
                     for col in item.get('columns',[]) or []:

src/viano_bug_analysis/analysis/investigation_prompt.py

@@ -15,6 +15,9 @@ def output_schema(context=None, *, strict=False):
         return schema(context,strict=strict)
     name = "investigation-project-output-v1.schema.json" if context and "x-project-bindings" in context["tool_schema"] else "investigation-output-v1.schema.json"
     schema=json.loads(files("viano_bug_analysis.schemas").joinpath(name).read_text())
+    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.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

@@ -25,6 +25,10 @@ def _result_status(capture):
             count=items[0].get('matched_rows')
             if type(count) is int and count>=0:
                 status['matched_rows']=count
+    elif query.get('kind')=='database.relations':
+        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 []
     # Unknown providers and selected source ranges never imply complete absence.
     return status
 

src/viano_bug_analysis/analysis/investigator.py

@@ -245,8 +245,17 @@ class Investigator:
             context["tool_schema"]["x-project-bindings"] = {
                 "source":service.source.binding() if service.source else None,
                 "database":service.database.binding() if service.database else None,
-                "query_schema":service.schema(), "max_result_bytes":service.limits.max_result_bytes,
+                "query_schema":service.query_schema(), "max_result_bytes":service.limits.max_result_bytes,
                 "max_rows":service.limits.max_rows, "max_lines":service.limits.max_lines}
+            if service.relations:
+                from viano_bug_analysis.tools.database_relations import extend_tool_schema
+                context['tool_schema']['x-project-bindings']['relation_profiles'] = service.relations.catalog()
+                context['tool_schema']['x-project-bindings']['relation_preference'] = (
+                    'Prefer a matching database.relations profile to separate schema/row/link calls. '
+                    'Use primitive tools for missing details. Complete means only configured edges; '
+                    'upstream truncation invalidates downstream absence claims.')
+                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"]=[
@@ -572,6 +581,8 @@ class Investigator:
         binding=context["tool_schema"].get("x-project-bindings",{})
         if service is None or binding.get("source")!=(service.source.binding() if service.source else None) or binding.get("database")!=(service.database.binding() if service.database else None):
             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')
 
     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

@@ -684,6 +684,31 @@
                         "minimum": 1
                       }
                     }
+                  },
+                  {
+                    "type": "object",
+                    "additionalProperties": false,
+                    "required": [
+                      "kind",
+                      "profile",
+                      "value"
+                    ],
+                    "properties": {
+                      "kind": {
+                        "const": "database.relations"
+                      },
+                      "profile": {
+                        "type": "string",
+                        "pattern": "^[A-Za-z_][A-Za-z0-9_]{0,63}$"
+                      },
+                      "value": {
+                        "type": [
+                          "string",
+                          "integer"
+                        ],
+                        "maxLength": 200
+                      }
+                    }
                   }
                 ]
               }

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

@@ -1685,6 +1685,31 @@
                                       "minimum": 1
                                     }
                                   }
+                                },
+                                {
+                                  "type": "object",
+                                  "additionalProperties": false,
+                                  "required": [
+                                    "kind",
+                                    "profile",
+                                    "value"
+                                  ],
+                                  "properties": {
+                                    "kind": {
+                                      "const": "database.relations"
+                                    },
+                                    "profile": {
+                                      "type": "string",
+                                      "pattern": "^[A-Za-z_][A-Za-z0-9_]{0,63}$"
+                                    },
+                                    "value": {
+                                      "type": [
+                                        "string",
+                                        "integer"
+                                      ],
+                                      "maxLength": 200
+                                    }
+                                  }
                                 }
                               ]
                             }

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

@@ -239,6 +239,31 @@
               "minimum": 1
             }
           }
+        },
+        {
+          "type": "object",
+          "additionalProperties": false,
+          "required": [
+            "kind",
+            "profile",
+            "value"
+          ],
+          "properties": {
+            "kind": {
+              "const": "database.relations"
+            },
+            "profile": {
+              "type": "string",
+              "pattern": "^[A-Za-z_][A-Za-z0-9_]{0,63}$"
+            },
+            "value": {
+              "type": [
+                "string",
+                "integer"
+              ],
+              "maxLength": 200
+            }
+          }
         }
       ]
     },
@@ -271,7 +296,9 @@
             "complete_file_inventory",
             "selected_lines",
             "partial_search",
-            "complete_selected_text_search"
+            "complete_selected_text_search",
+            "complete_configured_relations",
+            "partial_configured_relations"
           ]
         },
         "truncated": {

src/viano_bug_analysis/tools/database_relations.py

@@ -0,0 +1,173 @@
+"""Operator-defined, bounded relational plans; no inferred joins or diagnosis."""
+import copy
+import hashlib
+import json
+import re
+import time
+
+from viano_bug_analysis.lifecycle.storage import json_bytes
+from .config import ToolError
+
+
+def request_schema():
+    return {'type': 'object', 'additionalProperties': False,
+            'required': ['kind', 'profile', 'value'], 'properties': {
+                'kind': {'const': 'database.relations'},
+                'profile': {'type': 'string', 'pattern': '^[A-Za-z_][A-Za-z0-9_]{0,63}$'},
+                'value': {'type': ['string', 'integer'], 'maxLength': 200}}}
+
+
+def extend_query_schema(schema):
+    schema = copy.deepcopy(schema)
+    schema['oneOf'].append(request_schema())
+    return schema
+
+
+def extend_tool_schema(schema):
+    """Add the operation only to a new context with configured profiles."""
+    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
+
+
+class DatabaseRelations:
+    MAX_NODES = 12
+
+    def __init__(self, provider, profiles):
+        from .sql_database import identifier
+        self.provider = provider
+        self.profiles = copy.deepcopy(profiles)
+        if (not callable(getattr(provider, '_connect', None))
+                or not callable(getattr(provider, '_query_cursor', None))):
+            raise ToolError('PROJECT_RELATIONS_CONFIG')
+        if not isinstance(profiles, dict) or len(profiles) > 20:
+            raise ToolError('PROJECT_RELATIONS_CONFIG')
+        for name, profile in self.profiles.items():
+            identifier(name)
+            if (set(profile) != {'description', 'nodes'} or not isinstance(profile['description'], str)
+                    or not 1 <= len(profile['description']) <= 500
+                    or not isinstance(profile['nodes'], list) or not 1 <= len(profile['nodes']) <= self.MAX_NODES):
+                raise ToolError('PROJECT_RELATIONS_CONFIG')
+            seen = {}
+            for index, node in enumerate(profile['nodes']):
+                if not isinstance(node, dict) or set(node) - {'name', 'table', 'columns', 'match_column', 'parent', 'parent_column', 'group_by'}:
+                    raise ToolError('PROJECT_RELATIONS_CONFIG')
+                for key in ('name', 'table', 'match_column'):
+                    identifier(node[key])
+                if node['name'] in seen or node['table'] not in provider.tables:
+                    raise ToolError('PROJECT_RELATIONS_CONFIG')
+                columns = node['columns']
+                if (not isinstance(columns, list) or not 1 <= len(columns) <= 30
+                        or len(set(columns)) != len(columns)):
+                    raise ToolError('PROJECT_RELATIONS_CONFIG')
+                for col in columns: identifier(col)
+                if index == 0:
+                    if 'parent' in node or 'parent_column' in node:
+                        raise ToolError('PROJECT_RELATIONS_CONFIG')
+                elif (node.get('parent') not in seen
+                      or node.get('parent_column') not in seen[node['parent']]['columns']):
+                    raise ToolError('PROJECT_RELATIONS_CONFIG')
+                groups = node.get('group_by', [])
+                if not isinstance(groups, list) or len(groups) > 5 or len(set(groups)) != len(groups) or not set(groups) <= set(columns):
+                    raise ToolError('PROJECT_RELATIONS_CONFIG')
+                seen[node['name']] = node
+
+    def catalog(self):
+        return {name: {'description': p['description'],
+                       'nodes': [{k: copy.deepcopy(v) for k, v in n.items()} for n in p['nodes']]}
+                for name, p in self.profiles.items()}
+
+    def binding(self, profile):
+        return hashlib.sha256(json_bytes(self.profiles[profile])).hexdigest()
+
+    def validate(self, args):
+        if args['profile'] not in self.profiles:
+            raise ToolError('PROJECT_RELATIONS_PROFILE')
+        if type(args['value']) not in (str, int):
+            raise ToolError('PROJECT_QUERY_INTEGER')
+
+    def query(self, args):
+        self.validate(args)
+        provider = self.provider
+        profile = self.profiles[args['profile']]
+        deadline = time.monotonic() + provider.limits.timeout_seconds
+        connection = provider._connect()
+        results, parents = [], {}
+        try:
+            with connection.cursor() as cursor:
+                cursor.execute('SET SESSION max_statement_time = %s', (provider.limits.timeout_seconds,))
+                cursor.execute('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ')
+                cursor.execute('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY')
+                tables = sorted({n['table'] for n in profile['nodes']})
+                cursor.execute('SELECT TABLE_NAME, ENGINE FROM information_schema.TABLES '
+                               'WHERE TABLE_SCHEMA=%s AND TABLE_NAME IN (' + ','.join('%s' for _ in tables) + ')',
+                               (provider.database, *tables))
+                engines = {r['TABLE_NAME']: r['ENGINE'] for r in cursor.fetchall()}
+                if set(engines) != set(tables) or any(str(e).lower() not in {'innodb', 'xtradb'} for e in engines.values()):
+                    raise ToolError('PROJECT_RELATIONS_SNAPSHOT_UNSUPPORTED')
+                # Schema and rows are read through the same transaction/connection.
+                for node in profile['nodes']:
+                    if time.monotonic() >= deadline:
+                        raise ToolError('PROJECT_RELATIONS_TIMEOUT')
+                    cursor.execute('SET SESSION max_statement_time = %s', (max(0.001, deadline - time.monotonic()),))
+                    parent = parents.get(node.get('parent'))
+                    values = [args['value']] if parent is None else [r[node['parent_column']] for r in parent['rows']]
+                    values = list({json.dumps(v, sort_keys=True): v for v in values if v is not None}.values())
+                    inherited = parent is None or parent['complete']
+                    schema = provider._query_cursor(cursor, {'kind': 'database.schema', 'table': node['table']})
+                    columns = schema['items'][0]['columns']
+                    known = {c['COLUMN_NAME'] for c in columns}
+                    if not set(node['columns'] + [node['match_column']]) <= known:
+                        raise ToolError('PROJECT_DATABASE_COLUMN_UNKNOWN')
+                    if parent is None:
+                        typ = next(c['COLUMN_TYPE'] for c in columns if c['COLUMN_NAME'] == node['match_column'])
+                        if re.match(r'(tinyint|smallint|mediumint|int|bigint)\b', typ, re.I) and type(args['value']) is not int:
+                            raise ToolError('PROJECT_RELATIONS_VALUE_TYPE')
+                    # Parent pages have at most max_rows entries: no unbounded IN list.
+                    query = {'kind': 'database.rows', 'table': node['table'], 'columns': node['columns'],
+                             'filters': [{'column': node['match_column'], 'operator': 'in', 'value': values}],
+                             'offset': 0, 'limit': provider.limits.max_rows}
+                    if values:
+                        remaining = max(0.001, deadline - time.monotonic())
+                        cursor.execute('SET SESSION max_statement_time = %s', (remaining,))
+                        result = provider._query_cursor(cursor, query, schema=columns)
+                    else:
+                        result = {'items': [], 'coverage': 'complete_filtered_rows', 'truncated': False}
+                    rows = [item['row'] for item in result['items']]
+                    complete = inherited and not result['truncated']
+                    visible_columns = set(node['columns'] + [node['match_column']]) | {c['COLUMN_NAME'] for c in columns if c['COLUMN_KEY'] == 'PRI'}
+                    item = {'node': node['name'], 'table': node['table'], 'parent': node.get('parent'),
+                            'query': query if values else None,
+                            'skip_reason': None if values else 'no_nonnull_parent_keys',
+                            'columns': [c for c in columns if c['COLUMN_NAME'] in visible_columns],
+                            'schema_scope': 'projected_match_and_primary_key_columns_only', 'rows': rows,
+                            'returned_rows': len(rows), 'local_truncated': result['truncated'],
+                            'upstream_complete': inherited, 'complete': complete,
+                            'coverage': 'complete_configured_relation' if complete else 'partial_configured_relation'}
+                    if node.get('group_by'):
+                        counts = {}
+                        for row in rows:
+                            key = json.dumps([row[c] for c in node['group_by']], sort_keys=True)
+                            counts[key] = counts.get(key, 0) + 1
+                        item['distribution'] = {'columns': node['group_by'], 'scope': 'returned_rows_only',
+                                                'groups': [{'values': json.loads(k), 'count': v} for k, v in sorted(counts.items())]}
+                    results.append(item)
+                    parents[node['name']] = item
+                    if len(json_bytes(results)) > provider.limits.max_result_bytes:
+                        raise ToolError('PROJECT_RESULT_LIMIT')
+            partial = any(not r['complete'] for r in results)
+            if time.monotonic() >= deadline:
+                raise ToolError('PROJECT_RELATIONS_TIMEOUT')
+            return {'items': results, 'coverage': 'partial_configured_relations' if partial else 'complete_configured_relations',
+                    'truncated': partial,
+                    'interpretation': 'Only the named configured edges and projected columns were examined; '
+                                      'no claim about other tables, dynamic code, live server state or root cause.'}
+        except ToolError:
+            raise
+        except Exception:
+            raise ToolError('PROJECT_DATABASE_QUERY_FAILED') from None
+        finally:
+            try: connection.rollback()
+            finally: connection.close()

src/viano_bug_analysis/tools/project_query.py

@@ -35,11 +35,18 @@ class ProjectResearch:
     a URL, directory, credentials, database, branch, or arbitrary SQL.
     """
 
-    def __init__(self, project, *, source=None, database=None, limits=None):
+    def __init__(self, project, *, source=None, database=None, limits=None, relations=None):
         self.project = copy.deepcopy(project)
         self.source, self.database = source, database
         self.limits = limits or ProjectLimits()
         self._captures = OrderedDict()
+        from .database_relations import DatabaseRelations
+        if relations and database is None:
+            raise ToolError('PROJECT_RELATIONS_CONFIG')
+        try:
+            self.relations = DatabaseRelations(database, relations) if relations else None
+        except (ValueError, TypeError, KeyError):
+            raise ToolError('PROJECT_RELATIONS_CONFIG') from None
 
     @classmethod
     def load(cls, path):
@@ -48,7 +55,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"}:
+            if set(config) - {"project", "source", "database", "limits", "enrichment", "entity_resolution", "relations", "relations_file"}:
                 raise ValueError()
             limits = ProjectLimits(**config.get("limits", {}))
             source = None
@@ -60,7 +67,12 @@ class ProjectResearch:
             if db_settings.get("password_file"):
                 db_settings["password_file"] = (path.parent / db_settings["password_file"]).resolve()
             database = SQLDatabaseProvider(limits=limits, **db_settings) if db_settings else None
-            service = cls(config["project"], source=source, database=database, limits=limits)
+            if 'relations' in config and 'relations_file' in config:
+                raise ValueError()
+            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.enrichment = copy.deepcopy(config.get("enrichment", {}))
             service.entity_resolution = copy.deepcopy(config.get("entity_resolution", {}))
             return service
@@ -75,13 +87,17 @@ class ProjectResearch:
         if any(project.get(k) != self.project.get(k) for k in ("type", "profile")):
             raise ToolError("PROJECT_PROFILE_MISMATCH")
 
+    def query_schema(self):
+        from .database_relations import extend_query_schema
+        return extend_query_schema(self.schema()) if self.relations else self.schema()
+
     def prepare(self, context, request):
         self.check_project(context.reader.manifest["project"])
         self.validate(request["arguments"])
 
     def validate(self, arguments):
         from jsonschema import Draft202012Validator
-        if not Draft202012Validator(self.schema()).is_valid(arguments):
+        if not Draft202012Validator(self.query_schema()).is_valid(arguments):
             raise ToolError("PROJECT_QUERY_INVALID")
         # jsonschema accepts integral floats as integer; our contracts do not.
         for name in ("limit", "offset", "start_line", "end_line"):
@@ -92,17 +108,26 @@ class ProjectResearch:
         provider = self.database if arguments["kind"].startswith("database.") else self.source
         if provider is None:
             raise ToolError("PROJECT_PROVIDER_UNAVAILABLE")
+        if arguments['kind'] == 'database.relations':
+            self.relations.validate(arguments)
+            return
         provider.validate(arguments)
 
     def collect(self, arguments):
         self.validate(arguments)
         provider = self.database if arguments["kind"].startswith("database.") else self.source
         binding = provider.binding()
+        if arguments['kind'] == 'database.relations':
+            binding = {**binding, 'relation_profile': arguments['profile'],
+                       'consistency': 'single_repeatable_read_read_only_transaction',
+                       'relation_plan': copy.deepcopy(self.relations.profiles[arguments['profile']]),
+                       'relation_profile_sha256': self.relations.binding(arguments['profile'])}
         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 = provider.query(copy.deepcopy(arguments))
+            result = (self.relations.query(copy.deepcopy(arguments)) if arguments['kind'] == 'database.relations'
+                      else 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/router.py

@@ -43,7 +43,14 @@ class InvestigationToolRouter:
 
     @staticmethod
     def request_schema():
-        return json.loads(files("viano_bug_analysis.schemas").joinpath("investigation-request-v1.schema.json").read_text())
+        schema = json.loads(files("viano_bug_analysis.schemas").joinpath("investigation-request-v1.schema.json").read_text())
+        # Keep historical context/request hashes unchanged when profiles are absent.
+        for branch in schema['properties']['requests']['items']['oneOf']:
+            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']
+        return schema
 
     def query(self, package, batch):
         try:

src/viano_bug_analysis/tools/sql_database.py

@@ -142,35 +142,7 @@ class SQLDatabaseProvider:
             with connection.cursor() as cursor:
                 cursor.execute("SET SESSION max_statement_time = %s", (self.limits.timeout_seconds,))
                 cursor.execute("START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY")
-                # Metadata is constrained by the configured database/table, not arbitrary SQL.
-                cursor.execute("SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY FROM information_schema.COLUMNS "
-                               "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION",
-                               (self.database, args["table"]))
-                schema = cursor.fetchall()
-                names = [c["COLUMN_NAME"] for c in schema]
-                if not names:
-                    raise ToolError("PROJECT_DATABASE_TABLE_UNKNOWN")
-                if args["kind"] == "database.schema":
-                    return {"items":[{"table":args["table"], "columns":schema}], "coverage":"schema_only", "truncated":False}
-                requested = set(args["columns"]) | {f["column"] for f in args["filters"]}
-                if not requested <= set(names):
-                    raise ToolError("PROJECT_DATABASE_COLUMN_UNKNOWN")
-                columns = args["columns"] or names[:30]
-                where, values = self.compile_filters(args["filters"])
-                table = identifier(args["table"])
-                if args["kind"] == "database.count":
-                    cursor.execute("SELECT COUNT(*) AS matched_rows FROM " + table + " WHERE " + where, values)
-                    return {"items":cursor.fetchall(), "coverage":"complete_filtered_count", "truncated":False}
-                # Deterministic order uses unique key when possible, otherwise all selected columns.
-                order = [c["COLUMN_NAME"] for c in schema if c["COLUMN_KEY"] == "PRI"] or columns
-                sql = ("SELECT " + ",".join(identifier(c) for c in columns) + " FROM " + table + " WHERE " + where
-                       + " ORDER BY " + ",".join(identifier(c) for c in order) + " LIMIT %s OFFSET %s")
-                cursor.execute(sql, (*values, args["limit"]+1, args["offset"]))
-                rows = cursor.fetchall()
-                truncated = len(rows) > args["limit"]
-                return {"items":[{"table":args["table"], "row":{k:scalar(v) for k,v in row.items()}} for row in rows[:args["limit"]]],
-                        "coverage":"partial_filtered_rows" if truncated or args["offset"] else "complete_filtered_rows",
-                        "truncated":truncated, "next_offset":args["offset"]+args["limit"] if truncated else None}
+                return self._query_cursor(cursor, args)
         except ToolError:
             raise
         except Exception:
@@ -181,3 +153,35 @@ class SQLDatabaseProvider:
                 connection.rollback()
             finally:
                 connection.close()
+
+    def _query_cursor(self, cursor, args, *, schema=None):
+        # Metadata is constrained by the configured database/table, not arbitrary SQL.
+        if schema is None:
+            cursor.execute("SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY FROM information_schema.COLUMNS "
+                           "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION",
+                           (self.database, args["table"]))
+            schema = cursor.fetchall()
+        names = [c["COLUMN_NAME"] for c in schema]
+        if not names:
+            raise ToolError("PROJECT_DATABASE_TABLE_UNKNOWN")
+        if args["kind"] == "database.schema":
+            return {"items":[{"table":args["table"], "columns":schema}], "coverage":"schema_only", "truncated":False}
+        requested = set(args["columns"]) | {f["column"] for f in args["filters"]}
+        if not requested <= set(names):
+            raise ToolError("PROJECT_DATABASE_COLUMN_UNKNOWN")
+        columns = args["columns"] or names[:30]
+        where, values = self.compile_filters(args["filters"])
+        table = identifier(args["table"])
+        if args["kind"] == "database.count":
+            cursor.execute("SELECT COUNT(*) AS matched_rows FROM " + table + " WHERE " + where, values)
+            return {"items":cursor.fetchall(), "coverage":"complete_filtered_count", "truncated":False}
+        # Deterministic order uses unique key when possible, otherwise all selected columns.
+        order = [c["COLUMN_NAME"] for c in schema if c["COLUMN_KEY"] == "PRI"] or columns
+        sql = ("SELECT " + ",".join(identifier(c) for c in columns) + " FROM " + table + " WHERE " + where
+               + " ORDER BY " + ",".join(identifier(c) for c in order) + " LIMIT %s OFFSET %s")
+        cursor.execute(sql, (*values, args["limit"]+1, args["offset"]))
+        rows = cursor.fetchall()
+        truncated = len(rows) > args["limit"]
+        return {"items":[{"table":args["table"], "row":{k:scalar(v) for k,v in row.items()}} for row in rows[:args["limit"]]],
+                "coverage":"partial_filtered_rows" if truncated or args["offset"] else "complete_filtered_rows",
+                "truncated":truncated, "next_offset":args["offset"]+args["limit"] if truncated else None}

tests/test_database_relations.py

@@ -0,0 +1,264 @@
+"""Real SQL selection logic, simulated metadata/transaction transport; no LLM."""
+import copy
+import json
+import sqlite3
+from pathlib import Path
+import tomllib
+
+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.sql_database import SQLDatabaseProvider
+from viano_bug_analysis.tools.database_relations import DatabaseRelations
+from test_case_builder import case_setup
+from test_investigation_tools import tool_case
+
+
+PROFILE = {'inspect': {'description': 'Inspect parent records and equal-key children.', 'nodes': [
+    {'name': 'root', 'table': 'templates', 'match_column': 'entry', 'columns': ['entry', 'name']},
+    {'name': 'spawns', 'table': 'spawns', 'match_column': 'entry', 'parent': 'root',
+     'parent_column': 'entry', 'columns': ['guid', 'entry', 'zone', 'mask'], 'group_by': ['zone', 'mask']},
+    {'name': 'links', 'table': 'links', 'match_column': 'guid', 'parent': 'spawns',
+     'parent_column': 'guid', 'columns': ['guid', 'event']}]}}
+
+
+class Database:
+    def __init__(self):
+        self.db = sqlite3.connect(':memory:')
+        self.db.row_factory = sqlite3.Row
+        self.db.executescript('CREATE TABLE templates(entry INTEGER PRIMARY KEY, name TEXT);'
+                             'CREATE TABLE spawns(guid INTEGER PRIMARY KEY, entry INTEGER, zone INTEGER, mask INTEGER);'
+                             'CREATE TABLE links(guid INTEGER, event INTEGER);'
+                             "INSERT INTO templates VALUES (42,'subject');"
+                             'INSERT INTO spawns VALUES (1,42,14,1),(2,42,14,1),(3,42,14,1);'
+                             'INSERT INTO links VALUES (1,24),(2,24),(3,24);')
+        self.calls = []; self.connections = self.closed = self.rollbacks = 0
+        self.engine = 'InnoDB'; self.fail = False
+
+    def connect(self):
+        self.connections += 1
+        return self
+
+    def cursor(self): return self
+    def __enter__(self): return self
+    def __exit__(self, *args): pass
+    def close(self): self.closed += 1
+    def rollback(self): self.rollbacks += 1
+
+    def execute(self, sql, args=()):
+        self.calls.append((sql, args))
+        if self.fail: raise RuntimeError('private server error')
+        if sql.startswith(('SET ', 'START ')): self.result = []; return
+        if 'information_schema.TABLES' in sql:
+            self.result = [{'TABLE_NAME': t, 'ENGINE': self.engine} for t in args[1:]]; return
+        if 'information_schema.COLUMNS' in sql:
+            self.result = [{'COLUMN_NAME': r['name'], 'COLUMN_TYPE': 'int' if r['type']=='INTEGER' else 'text',
+                            'IS_NULLABLE': 'YES', 'COLUMN_KEY': 'PRI' if r['pk'] else ''}
+                           for r in self.db.execute('PRAGMA table_info(`'+args[1]+'`)')]; return
+        self.result = [dict(r) for r in self.db.execute(sql.replace('%s','?'), args)]
+
+    def fetchall(self): return copy.deepcopy(self.result)
+
+
+def service(limit=50, profile=None):
+    db = Database(); limits = ProjectLimits(max_rows=limit)
+    provider = SQLDatabaseProvider('localhost', 3306, 'fixture', 'reader', 'snapshot',
+        ['templates', 'spawns', 'links'], password_env='UNUSED', limits=limits, connection_factory=db.connect)
+    return ProjectResearch({}, database=provider, limits=limits, relations=profile or PROFILE), db
+
+
+def query(value=42): return {'kind': 'database.relations', 'profile': 'inspect', 'value': value}
+
+
+def test_complete_plan_single_readonly_snapshot_and_frozen_capture():
+    s, db = service()
+    capture = json.loads(s.collect(query()))
+    result = capture['result']; nodes = result['items']
+    assert result['coverage']=='complete_configured_relations' and not result['truncated']
+    assert [n['returned_rows'] for n in nodes]==[1,3,3]
+    assert nodes[1]['distribution']['groups']==[{'values':[14,1], 'count':3}]
+    assert nodes[2]['query']['filters'][0]['value']==[1,2,3]
+    assert capture['origin']['relation_profile_sha256']==s.relations.binding('inspect')
+    assert db.connections==db.closed==db.rollbacks==1
+    assert sum(sql.startswith('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY') for sql,_ in db.calls)==1
+    assert all(sql.startswith(('SELECT ', 'SET SESSION ', 'SET TRANSACTION ', 'START TRANSACTION ')) for sql,_ in db.calls)
+    before = len(db.calls)
+    assert json.loads(s.collect(query()))==capture and len(db.calls)==before
+
+
+def test_partial_parent_propagates_even_when_child_page_itself_is_complete():
+    s, db = service(limit=2)
+    # Remove one selected child's link so local child page is not truncated.
+    db.db.execute('DELETE FROM links WHERE guid=2')
+    r = json.loads(s.collect(query()))['result']; root,spawns,links=r['items']
+    assert root['complete'] and spawns['local_truncated']
+    assert not links['local_truncated'] and not links['upstream_complete'] and not links['complete']
+    assert r['truncated'] and r['coverage']=='partial_configured_relations'
+    assert spawns['distribution']['scope']=='returned_rows_only'
+
+
+def test_empty_root_and_null_key_skip_without_empty_in_or_global_absence():
+    s, db = service()
+    r=json.loads(s.collect(query(99)))['result']
+    assert [n['rows'] for n in r['items']]==[[],[],[]]
+    assert r['items'][1]['skip_reason']=='no_nonnull_parent_keys'
+    assert r['items'][1]['query'] is None
+    assert 'dynamic code' in r['interpretation']
+    assert not any('IN ()' in sql for sql,_ in db.calls)
+
+
+@pytest.mark.parametrize('value',["42 OR 1=1", "x' UNION SELECT * FROM links --"])
+def test_values_are_bound_not_sql(value):
+    profile=copy.deepcopy(PROFILE);profile['inspect']['nodes'][0]['match_column']='name'
+    s, db=service(profile=profile); assert json.loads(s.collect(query(value)))['result']['items'][0]['rows']==[]
+    assert all(value not in sql for sql,_ in db.calls)
+
+
+def test_numeric_start_keys_do_not_use_mysql_implicit_string_coercion():
+    s,_=service()
+    with pytest.raises(ToolError,match='VALUE_TYPE'):s.collect(query('42garbage'))
+
+
+@pytest.mark.parametrize('change',[{'profile':'absent'}, {'sql':'SELECT * FROM accounts'}, {'value':True}, {'value':42.0}, {'table':'accounts'}])
+def test_invalid_requests_do_not_connect(change):
+    s,db=service()
+    with pytest.raises(ToolError):s.collect({**query(),**change})
+    assert db.connections==0
+
+
+@pytest.mark.parametrize('mutation',['cycle','unknown_parent_column','forbidden_table','bad_identifier','too_many_nodes'])
+def test_invalid_operator_plans_are_rejected(mutation):
+    p=copy.deepcopy(PROFILE); n=p['inspect']['nodes']
+    if mutation=='cycle':n[1]['parent']='spawns'
+    elif mutation=='unknown_parent_column':n[1]['parent_column']='missing'
+    elif mutation=='forbidden_table':n[1]['table']='accounts'
+    elif mutation=='bad_identifier':n[1]['match_column']='x; DROP TABLE x'
+    else:p['inspect']['nodes']=n*5
+    with pytest.raises(ToolError):service(profile=p)
+
+
+def test_unknown_schema_engine_and_sql_errors_are_controlled():
+    s,db=service();s.relations.profiles['inspect']['nodes'][0]['columns']=['missing']
+    with pytest.raises(ToolError,match='COLUMN_UNKNOWN'):s.collect(query())
+    assert db.closed==db.rollbacks==1
+    s,db=service();db.engine='MyISAM'
+    with pytest.raises(ToolError,match='SNAPSHOT_UNSUPPORTED'):s.collect(query())
+    s,db=service();db.fail=True
+    with pytest.raises(ToolError,match='QUERY_FAILED') as e:s.collect(query())
+    assert 'private' not in str(e.value) and db.closed==1
+
+
+def test_result_limit_is_enforced_and_no_capture_published():
+    s,db=service();s.database.limits=ProjectLimits(max_result_bytes=100)
+    with pytest.raises(ToolError,match='RESULT_LIMIT'):s.collect(query())
+    assert not s._captures and db.closed==db.rollbacks==1
+
+
+def test_profile_change_changes_capture_binding_and_missing_profiles_stay_unadvertised():
+    s,db=service();a=s.collect(query())
+    s.relations.profiles['inspect']['description']='Updated meaning.'
+    b=s.collect(query());assert a!=b and db.connections==2
+    assert len(ProjectResearch.schema()['oneOf'])+1==len(s.query_schema()['oneOf'])
+    with pytest.raises(ToolError):ProjectResearch({},database=s.database).collect(query())
+
+
+def test_vcore_profile_is_adapter_data_and_generic_plan_has_no_wow_assumptions():
+    p=tomllib.loads(Path('config/project-relations.vcore.toml').read_text(encoding='utf-8'))
+    assert 'creature_spawn_context' in p['relations']
+    s,_=service(); assert 'Wolpertinger' not in json.dumps(s.relations.catalog())
+
+
+def test_new_context_strict_actions_allow_relations_but_old_context_stays_frozen():
+    from test_investigation_lean import context as lean_context
+    from viano_bug_analysis.analysis.investigation_prompt import output_schema
+    c=lean_context();c.pop('lean_session',None)
+    old=output_schema(c,strict=True)['properties']['next']
+    action={'kind':'tool','operation':'project.query','arguments':query()}
+    assert not Draft202012Validator(old).is_valid(action)
+    c['tool_schema']['x-project-bindings']['relation_profiles']=PROFILE
+    assert Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
+
+
+def test_total_deadline_is_controlled_and_publishes_no_partial_capture(monkeypatch):
+    import viano_bug_analysis.tools.database_relations as module
+    s,db=service();clock=iter([0,100])
+    monkeypatch.setattr(module.time,'monotonic',lambda:next(clock))
+    with pytest.raises(ToolError,match='RELATIONS_TIMEOUT'):s.collect(query())
+    assert not s._captures and db.closed==db.rollbacks==1
+
+
+def test_relational_rows_participate_in_coding_value_guard():
+    from viano_bug_analysis.analysis.database_value_grounding import unsupported_assignments
+    s,_=service();capture=json.loads(s.collect(query()))
+    c={'evidence':{'E000001':{'content':{'items':[capture]}}}}
+    assert unsupported_assignments({'implementation_guidance':['Set mask=14.']},c)==['mask']
+    assert unsupported_assignments({'implementation_guidance':['Set mask=1.']},c)==[]
+
+
+def test_relations_remain_portable_budgeted_evidence(tool_case,tmp_path):
+    from test_investigation_tools import batch
+    from viano_bug_analysis.tools import InvestigationToolRouter
+    from viano_bug_analysis.package import PackageReader,PackageValidator
+    from viano_bug_analysis.budget import BudgetManager
+    from viano_bug_analysis.evidence import EvidenceStore
+    from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
+    store=tool_case[0];s,db=service()
+    with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project']
+    router=InvestigationToolRouter(project_research=s)
+    budget=BudgetManager(store);budget.initialize('case-A','stable-package')
+    reserved=budget.reserve('case-A','stable-package','H1','A1','project.query')
+    asked=batch('project.query',query());reply=router.query(reserved.package,asked)
+    saved=EvidenceStore(store,router=router).store_tool_results('case-A','stable-package','relations-1',asked,reply)
+    assert PackageValidator().validate(saved.package).valid
+    budget.record_evidence('case-A','stable-package','H1','A1',saved.record_refs[0])
+    portable=CheckpointStore(Workspace(tmp_path/'portable'))
+    portable.intake(store.latest('case-A','stable-package'))
+    with PackageReader(portable.latest('case-A','stable-package')) as reader:
+        record=json.loads(reader.read_bytes(saved.record_refs[0]))
+        content=json.loads(reader.read_bytes(record['content_ref']))
+        capture=content['items'][0]
+        assert capture['result']['items'][2]['returned_rows']==3
+        assert capture['origin']['relation_plan']['nodes'][2]['parent_column']=='guid'
+    assert db.connections==1
+
+
+def test_hybrid_loop_uses_relations_with_fixture_llm_and_can_resume(tool_case,tmp_path):
+    from test_case_interpreter import gateway
+    from test_investigation_hybrid import HybridAdapter
+    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 Workspace,CheckpointStore
+    s,db=service()
+    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 'relation_profiles' in adapter.calls[0]['input']['tool_schema']['project']
+    portable=CheckpointStore(Workspace(tmp_path/'resumed'));portable.intake(first.package)
+    final=run(Investigator(portable,gateway(adapter),router=router))
+    assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
+    # Identical structured query is frozen once in the shared research service.
+    assert db.connections==1
+
+
+def test_real_local_world_relations():
+    import os
+    config=os.environ.get('VBA_TEST_PROJECT_CONFIG')
+    if not config:pytest.skip('Opt-in local SELECT-only database')
+    s=ProjectResearch.load(config)
+    capture=json.loads(s.collect({'kind':'database.relations','profile':'creature_spawn_context','value':23487}))
+    nodes={n['node']:n for n in capture['result']['items']}
+    assert not capture['result']['truncated']
+    assert nodes['template']['rows'][0]['entry']==23487
+    assert len(nodes['spawns']['rows'])==18
+    assert {r['map'] for r in nodes['spawns']['rows']}=={1}
+    assert {r['guid'] for r in nodes['event_links']['rows']}=={r['guid'] for r in nodes['spawns']['rows']}
+    assert {r['eventEntry'] for r in nodes['event_links']['rows']}=={24}
+    assert nodes['events']['rows'][0]['eventEntry']==24

tests/test_project_research.py

@@ -229,6 +229,13 @@ def test_model_loop_uses_project_tools_and_compact_previous_result(tool_case, re
             return ProviderResponse(reply,"fixture",100,10)
     adapter=Adapter()
     loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
+    if not recall:
+        original_context=loop._context
+        def frozen_v15_context(*args):
+            context=original_context(*args)
+            context['prompt_version']='investigation_project_v15'
+            return context
+        loop._context=frozen_v15_context
     if recall:
         original_context=loop._context
         def frozen_context(*args):
@@ -248,6 +255,12 @@ def test_model_loop_uses_project_tools_and_compact_previous_result(tool_case, re
             context['recalled_evidence']=args[0].get('recalled_evidence',{}).get(args[1],[])
             return context
         if recall is True:loop._context=frozen_resume_context
+        else:
+            def frozen_v15_resume(*args):
+                context=original_context(*args)
+                context['prompt_version']='investigation_project_v15'
+                return context
+            loop._context=frozen_v15_resume
     result=asyncio.run(loop.run("case-A","stable-package",max_steps=20))
     if recall == 'unknown':
         assert result.status == 'BLOCKED' and result.code == 'INVESTIGATION_EVIDENCE_UNKNOWN'