96 lines
5.6 KiB
Python
96 lines
5.6 KiB
Python
"""Necessary same-column grounding for explicit numeric DB assignments in prose.
|
|
|
|
This is a bounded guard, not a proof that an observed value suits a new entity.
|
|
It never substitutes values or modifies model instructions.
|
|
"""
|
|
import re
|
|
from decimal import Decimal
|
|
from math import isfinite
|
|
|
|
|
|
def unsupported_assignments(payload,context):
|
|
observed={}
|
|
table_columns={}
|
|
def remember(column,value):
|
|
if (isinstance(column,str) and re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]{0,127}',column)
|
|
and type(value) in (int,float) and (type(value) is int or isfinite(value))):
|
|
observed.setdefault(column.lower(),set()).add(Decimal(str(value)))
|
|
for entry in context.get('evidence',{}).values():
|
|
for capture in entry['content'].get('items',[]):
|
|
if not isinstance(capture,dict):continue
|
|
query=capture.get('query');result=capture.get('result')
|
|
if not isinstance(query,dict) or not isinstance(result,dict):continue
|
|
kind=query.get('kind')
|
|
if context.get('parameter_grounding') == 2 and kind in {'database.rows','database.schema','database.relations'}:
|
|
for item in result.get('items', []):
|
|
if not isinstance(item, dict): continue
|
|
table = item.get('table') or query.get('table')
|
|
if not isinstance(table, str): continue
|
|
columns = table_columns.setdefault(table, set())
|
|
for col in item.get('columns', []) or []:
|
|
if (isinstance(col, dict) and isinstance(col.get('COLUMN_NAME'), str)
|
|
and isinstance(col.get('COLUMN_TYPE'), str)
|
|
and re.match(r'(?:tinyint|smallint|mediumint|int|bigint|decimal|numeric|float|double)\b', col.get('COLUMN_TYPE', ''), re.I)):
|
|
columns.add(col['COLUMN_NAME'])
|
|
rows = item.get('rows', []) if kind == 'database.relations' else [item.get('row', {})]
|
|
for row in rows:
|
|
if isinstance(row, dict):
|
|
columns.update(k for k, v in row.items() if type(v) in (int, float))
|
|
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 []:
|
|
if not isinstance(col,dict):continue
|
|
name=col.get('COLUMN_NAME','');typ=col.get('COLUMN_TYPE','')
|
|
if isinstance(name,str) and isinstance(typ,str) and re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]{0,127}',name) and re.match(r'(?:tinyint|smallint|mediumint|int|bigint|decimal|numeric|float|double)\b',typ,re.I):
|
|
observed.setdefault(name.lower(),set())
|
|
if kind!='database.rows':continue
|
|
for item in capture.get('result',{}).get('items',[]):
|
|
if isinstance(item,dict) and isinstance(item.get('row'),dict):
|
|
for column,value in item['row'].items():remember(column,value)
|
|
# Named, typed facts in the canonical case can also establish a value.
|
|
def case_values(v):
|
|
if isinstance(v,dict):
|
|
for column,value in v.items():
|
|
if isinstance(column,str) and column.lower() in observed:remember(column,value)
|
|
case_values(value)
|
|
elif isinstance(v,list):
|
|
for item in v:case_values(item)
|
|
case_values(context.get('case',{}))
|
|
if context.get('parameter_grounding') == 2:
|
|
facts = context.get('case', {}).get('runtime_facts', {}).get('facts', [])
|
|
counts = {}
|
|
for fact in facts:
|
|
fid = fact.get('id')
|
|
if isinstance(fid, str): counts[fid] = counts.get(fid, 0) + 1
|
|
unique = {f['id']: f for f in facts if counts.get(f.get('id')) == 1}
|
|
for fact in unique.values():
|
|
# A named canonical fact is not a free-text model claim.
|
|
name = fact.get('name')
|
|
if isinstance(name, str) and name.lower() in observed:
|
|
remember(name, fact.get('value'))
|
|
from viano_bug_analysis.profiles.database_bindings import fact_bindings
|
|
# Re-derive adapter bindings; arbitrary descriptors cannot authorize values.
|
|
allowed = fact_bindings(context.get('case', {}))
|
|
for binding in context.get('database_fact_bindings', []):
|
|
if binding not in allowed: continue
|
|
fact = unique.get(binding['fact_id'])
|
|
if (fact and fact.get('name') == binding['fact_name']
|
|
and binding['column'] in table_columns.get(binding['table'], set())):
|
|
remember(binding['column'], fact.get('value'))
|
|
bad=[]
|
|
for text in payload.get('implementation_guidance',[]):
|
|
if not isinstance(text,str):continue
|
|
for column,values in observed.items():
|
|
# Schema-only names can be ordinary prose (e.g. 'entry 123').
|
|
# Require explicit assignment syntax until row values establish it.
|
|
separator=r'(?:\s*(?:=|:)\s*|\s+(?:is\s+)?)' if values else r'\s*(?:=|:)\s*'
|
|
pattern=r'(?<![A-Za-z0-9_])`?'+re.escape(column)+r'`?'+separator+r'(-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)(?![A-Za-z0-9_]|\.[0-9])'
|
|
for match in re.finditer(pattern,text,re.I):
|
|
if Decimal(match.group(1)) not in values:bad.append(column)
|
|
return sorted(set(bad))
|