61 lines
4.1 KiB
Python
61 lines
4.1 KiB
Python
"""Authorized read-only app/workflow configuration check for captured local runs."""
|
|
import hashlib
|
|
import json
|
|
import shlex
|
|
|
|
def inspect(client,base,root,state):
|
|
selected=[]
|
|
for profile,directory in [('qwen','server/runtime/adminapi/assistant-requests'),('openai','server/runtime/ai-synthetic-runtime-20261027/controls')]:
|
|
candidates=[]
|
|
for p in (root/directory).glob('*.json'):
|
|
s=json.loads(p.read_text(encoding='utf-8'))
|
|
if s.get('upstream',{}).get('profile')==profile:
|
|
candidates.append((p.stat().st_mtime,s))
|
|
base._require(bool(candidates),'CAPTURED_PROFILE_RUN_REQUIRED')
|
|
s=max(candidates,key=lambda v:v[0])[1]
|
|
selected.append({'profile':profile,'run_id':s['upstream']['workflow_run_id']})
|
|
expected=(root/'artifacts/ai_full_verification_20261027/dify_consultation_system_v2.txt').read_text(encoding='utf-8')
|
|
code='''import hashlib,json,os,re,sys
|
|
from sqlalchemy import create_engine,text
|
|
from sqlalchemy.engine import URL
|
|
stage='connect'
|
|
items=json.loads(sys.argv[1]); expected_hash=sys.argv[2]
|
|
kind=os.getenv('DB_TYPE','postgresql'); postgres=kind in ['postgresql','postgres']
|
|
driver='postgresql+psycopg2' if postgres else 'mysql+pymysql'
|
|
url=URL.create(driver,username=os.environ['DB_USERNAME'],password=os.environ['DB_PASSWORD'],host=os.environ['DB_HOST'],port=int(os.getenv('DB_PORT','5432' if postgres else '3306')),database=os.environ['DB_DATABASE'])
|
|
engine=create_engine(url,pool_pre_ping=False,connect_args={'connect_timeout':5})
|
|
results=[]
|
|
with engine.connect() as c:
|
|
with c.begin():
|
|
stage='read_only_transaction'
|
|
c.exec_driver_sql('SET TRANSACTION READ ONLY')
|
|
for item in items:
|
|
stage='run_configuration_query'
|
|
assert re.fullmatch('[a-zA-Z0-9_-]{1,128}',item['run_id'])
|
|
run=c.execute(text('SELECT app_id,workflow_id FROM workflow_runs WHERE id=:id'),{'id':item['run_id']}).mappings().one()
|
|
stage='app_configuration_query'
|
|
app=c.execute(text('SELECT workflow_id FROM apps WHERE id=:id'),{'id':run['app_id']}).mappings().one()
|
|
stage='workflow_configuration_query'
|
|
w=c.execute(text('SELECT id,version,graph FROM workflows WHERE id=:id'),{'id':run['workflow_id']}).mappings().one()
|
|
stage='workflow_graph_fingerprint'
|
|
graph=json.loads(w['graph']) if isinstance(w['graph'],str) else w['graph']; nodes=[]
|
|
for node in graph.get('nodes',[]):
|
|
data=node.get('data',{})
|
|
if data.get('type')!='llm':continue
|
|
systems=[v.get('text','') for v in data.get('prompt_template',[]) if v.get('role')=='system']
|
|
prompts=[]
|
|
for value in systems:
|
|
normalized=value.replace('\\r\\n','\\n')
|
|
prompts.append({'characters':len(value),'sha256':hashlib.sha256(normalized.encode()).hexdigest(),'v2_exact':hashlib.sha256(normalized.encode()).hexdigest()==expected_hash,'scope_marker_present':'[TCM_CLINICAL_RESPONSE:tcm-consultation-task-analysis-r37]' in value,'v2_branch_sentence_present':'正文只省略脉象栏目，不描述脉象缺项、补采或采集流程' in value})
|
|
nodes.append({'node_id':node.get('id'),'systems':prompts})
|
|
results.append({'profile':item['profile'],'app_id':str(run['app_id']),'workflow_id':str(w['id']),'workflow_version':str(w['version']),'matches_current_published':app['workflow_id']==run['workflow_id'],'llm_nodes':nodes})
|
|
engine.dispose()
|
|
stage='safe_export'
|
|
print(json.dumps({'ok':True,'results':results,'database_read_only_transaction':True,'raw_prompts_or_inputs_exported':False}))
|
|
'''
|
|
protected='import json\nstage="imports"\ntry:\n'+'\n'.join(' '+line for line in code.splitlines())+'\nexcept Exception as e:\n print(json.dumps({"ok":False,"stage":stage,"error_type":type(e).__name__}))\n'
|
|
compile(protected,'<read-only-config-check>','exec')
|
|
answer,unused=base._json_command(client,'docker exec '+shlex.quote(state['deployed_container_id'])+' python -B -c '+shlex.quote(protected)+' '+shlex.quote(json.dumps(selected))+' '+shlex.quote(hashlib.sha256(expected.encode()).hexdigest()),timeout=25)
|
|
(root/'artifacts/ai_full_verification_20261027/published_runtime_identity.safe.json').write_text(json.dumps(answer,indent=2)+'\n',encoding='utf-8')
|
|
return answer
|