#!/usr/bin/env python3
"""Deterministic post-session knowledge review.
Scans all local Hermes profile state DBs, extracts only explicit durable facts,
and writes provenance-bearing project notes. No LLM, no transcript archive.
"""
from __future__ import annotations
import argparse, hashlib, json, sqlite3, time, re, unicodedata, subprocess
from pathlib import Path
ROOT=Path('/home/hermes/.hermes'); VAULT=Path('/opt/obsidian-vault')
STATE=Path('/opt/struktur/session-knowledge-review/review.db'); AUDIT=Path('/opt/struktur/session-knowledge-review/audit.jsonl')
VERSION='session-knowledge-review/1.2.0'
def db():
STATE.parent.mkdir(parents=True,exist_ok=True)
c=sqlite3.connect(STATE);c.row_factory=sqlite3.Row;c.execute('''create table if not exists session_reviews (session_key text primary key, session_id text, profile text, source_db text, project text, status text not null, knowledge_items_found integer not null default 0, created integer not null, completed_at real, error text, version text not null)''');c.execute('''create table if not exists semantic_reviews (session_key text primary key, session_id text, profile text, project text, status text not null, knowledge_items_found integer not null default 0, created_at real not null, completed_at real, version text not null, error text)''');c.execute('''create table if not exists semantic_items (session_key text, claim_id text, project text, subject text, claim text, claim_type text, source_session_id text, source_profile text, source_timestamp real, confidence real, validation_status text, knowledge_action text, target text, primary key(session_key,claim_id))''');c.execute('''create table if not exists agent_reviews (item_key text primary key, semantic_item_id text not null, claim_id text, project text, source_session_id text, source_profile text, review_agent text not null, review_agent_version text not null, reviewed_at real not null, decision text, decision_reason text, decision_confidence real, fact_check_required integer not null default 0, fact_check_status text, final_action text, error text)''');c.commit();return c
def project_for(title,text):
# Classification is title/auftrag-first. Do not let inherited system prompts,
# quoted handovers, or a generic word in the transcript override the actual task.
t=re.sub(r'[-_]+',' ',(title or '').lower())
if any(k in t for k in ('mas visual library','visual kurator','asset group','variantenlogik','variant candidate')):
return 'MAS Visual Library'
if any(k in t for k in ('social media radar','social media agent','sma radar','aa 045','facebook property','social media')):
return 'Social Media Radar'
if 'youtube lm' in t or 'youtube video' in t or 'youtube audio' in t:return 'YouTube LM'
if 'youtube research' in t or 'transkript' in t:return 'YouTube Research'
if 'vento' in t:return 'Vento'
if any(k in t for k in ('worker watcher','wachfunktion','überwachung')):return 'Worker-Überwachung'
# Body fallback only accepts direct, project-specific production identifiers.
body=(text or '')[:20000].lower()
if '/opt/struktur/mas-visual-library/' in body or 'visual_library.db' in body:return 'MAS Visual Library'
return 'MAS allgemein'
def claim(cid,statement,confidence,source_type='session+live_validation'):
return {'claim_id':cid,'statement':statement,'confidence':confidence,'source_type':source_type}
def semantic_extract(project,sid,title,text):
"""Extract explicit, stable claims; this is claim extraction, not summarization.
Only user-authored durable markers and verifiable production paths qualify.
"""
out=[]; seen=set(); body='\n'.join(x.strip() for x in (text or '').splitlines() if x.strip())
for line in body.splitlines():
low=line.lower()
if '/tmp/' in low or 'debug' in low or re.search(r'\b(job|run|session)[ _-]?id\b',low): continue
m=re.search(r'(/opt/(?:struktur|obsidian-vault)/[^\s`\"\']+)',line)
if m and len(line)<=180 and not line.lstrip().startswith(('-', '*', '•')) and not any(k in low for k in ('reporter:','status=','bitte','arbeitsauftrag','noch offen','offen:','verifiziert unter','prüfe','pruefe','soll','muss')) and (('liegt' in low) or ('pfad' in low) or ('datenbank' in low) or ('vault' in low)) and re.match(r'^(die|der|das|hermes|obsidian)\b',low):
path=m.group(1).rstrip('.,)'); key='path.'+hashlib.sha256(path.encode()).hexdigest()[:12]
if key not in seen:
seen.add(key);out.append({'claim_id':f'{project.lower().replace(" ",".")}.{key}','subject':'production_path','claim':line,'statement':line,'claim_type':'PATH','confidence':0.97,'validation_status':'VALIDATED' if Path(path).exists() else 'SUPPORTED_BY_SESSION','source_type':'session_user_instruction'})
# A work-order imperative is not a fact. Accept only declarative markers.
if re.match(r'^(die|der|das|hermes|obsidian|der aktuelle|die aktuelle)\b',low) and any(k in low for k in (' ist ',' sind ',' liegt ',' bleibt ',' dient ',' erfolgt ',' wurde ',' gilt:')) and '?' not in low and not re.match(r'^(prüfe|pruefe|erstelle|lege|führe|fuehre|behalte|muss|soll|darf|implementiere|aktualisiere|bitte|vor jeder|für |fuer )\b',low):
clean=re.sub(r'\s+',' ',line).strip()
if len(clean)>=35 and len(clean)<=300 and not any(k in low for k in ('noch ','unter anderem','nicht als','häufig','beispielsweise','arbeitsauftrag','status=','prüfe','pruefe','sollte','könnte','koennte','bitte')) and not clean.endswith(':'):
slug=re.sub(r'[^a-z0-9]+','_',unicodedata.normalize('NFKD',clean.lower())).strip('_')[:70]
key=f'{project.lower().replace(" ",".")}.rule.{slug}'
if key not in seen:
seen.add(key);out.append({'claim_id':key,'subject':'explicit_rule','claim':clean,'statement':clean,'claim_type':'RULE','confidence':0.86,'validation_status':'SUPPORTED_BY_SESSION','source_type':'session_user_instruction','knowledge_action':'REVIEW_REQUIRED'})
return out
def semantic_one(c,profile,dbpath,row):
sid=row['id']; key=f'{profile}:{sid}'
if c.execute('select 1 from semantic_reviews where session_key=?',(key,)).fetchone(): return None
q=sqlite3.connect('file:'+dbpath+'?mode=ro',uri=True); text='\n'.join(x[0] or '' for x in q.execute("select content from messages where session_id=? and role='user' order by id",(sid,)))
project=project_for(row['title'] or '',text); items=semantic_extract(project,sid,row['title'] or '',text); result={'event':'semantic_review','session_id':sid,'project':project,'items':[],'status':'COMPLETED','timestamp':time.time()}
try:
if not VAULT.exists(): raise RuntimeError('OBSIDIAN_VAULT_UNAVAILABLE')
for item in items:
# Low-confidence claims are recorded for review, never written as durable truth.
if item.get('confidence',0)<.90:
item['knowledge_action']='REVIEW_REQUIRED'; result['items'].append(item); continue
action,path=append_note(note_path(project),project,item,{'session_id':sid,'profile':profile});item['knowledge_action']=action;item['target']=str(path);result['items'].append(item)
status='REVIEW_REQUIRED' if any(x.get('knowledge_action')=='REVIEW_REQUIRED' for x in result['items']) else 'COMPLETED'
except Exception as e: status='KNOWLEDGE_REVIEW_PENDING';result['status']=status;result['error']=str(e)
for item in result['items']:
c.execute('insert or replace into semantic_items values(?,?,?,?,?,?,?,?,?,?,?,?,?)',(key,item.get('claim_id'),project,item.get('subject'),item.get('claim'),item.get('claim_type'),sid,profile,time.time(),item.get('confidence'),item.get('validation_status'),item.get('knowledge_action'),item.get('target')))
result['status']=status
c.execute('insert into semantic_reviews values(?,?,?,?,?,?,?,?,?,?)',(key,sid,profile,project,status,len(items),time.time(),time.time() if status=='COMPLETED' else None,VERSION,result.get('error')));c.commit();
AUDIT.parent.mkdir(parents=True,exist_ok=True);AUDIT.open('a',encoding='utf-8').write(json.dumps(result,ensure_ascii=False)+'\n');return result
def _agent_json(text):
for i,ch in enumerate(text or ''):
if ch=='{':
try:
x=json.loads(text[i:text.rfind('}')+1]);
if isinstance(x,dict) and x.get('decision') in {'CREATE','CONFIRM','CONFLICT','REJECT'}: return x
except Exception: pass
return None
def run_wissenspruefer(item):
payload={'semantic_item_id':f"{item['session_key']}:{item['claim_id']}",'claim_id':item['claim_id'],'project':item['project'],'subject':item['subject'],'claim':item['claim'],'claim_type':item['claim_type'],'source_session_id':item['source_session_id'],'source_profile':item['source_profile'],'source_timestamp':item['source_timestamp'],'confidence':item['confidence'],'validation_status':item['validation_status'],'existing_obsidian_claim':None,'technical_validation':item['validation_status'],'relevant_context':item['claim']}
prompt=('Prüfe ausschließlich diesen einzelnen REVIEW_REQUIRED-Wissensclaim. Keine Zusammenfassung. '
'Bestätigte produktive Pfade, Datenmodelle, Rollen, Konfigurationen, Reparaturen und Prozessregeln sind dauerhaft relevant; Aktualität allein ist kein Ablehnungsgrund. Temporäre Pfade, Arbeitsaufträge, Zwischenstände und Spekulationen sind REJECT. '
'Antworte ausschließlich als JSON mit decision CREATE|CONFIRM|CONFLICT|REJECT, reason, confidence, needs_fact_check boolean, fact_check_reason. Claim: '+json.dumps(payload,ensure_ascii=False))
cmd=['sudo','-u','hermes','env','HOME=/home/hermes','USER=hermes','LOGNAME=hermes','/home/hermes/.local/bin/hermes','-p','wissenspruefer','chat','--oneshot','-q',prompt]
try:
p=subprocess.run(cmd,text=True,capture_output=True,timeout=180); d=_agent_json(p.stdout)
if not d: return {'error':'invalid_agent_json','raw':p.stdout[-1000:]}
return d
except Exception as e:return {'error':str(e)}
def run_faktenpruefer(item):
payload={'claim_id':item['claim_id'],'project':item['project'],'claim':item['claim'],'validation_status':item['validation_status'],'source_session_id':item['source_session_id']}
prompt=('Prüfe ausschließlich die Belegbarkeit dieses Claims. Antworte ausschließlich als JSON mit verdict VALIDATED|UNVERIFIED|CONFLICT|REJECT, reason und confidence. Keine Zusammenfassung, keine neue Tatsache. Claim: '+json.dumps(payload,ensure_ascii=False))
cmd=['sudo','-u','hermes','env','HOME=/home/hermes','USER=hermes','LOGNAME=hermes','/home/hermes/.local/bin/hermes','-p','faktenpruefer','chat','--oneshot','-q',prompt]
try:
p=subprocess.run(cmd,text=True,capture_output=True,timeout=180)
for i,ch in enumerate(p.stdout or ''):
if ch=='{':
try:
x=json.loads(p.stdout[i:p.stdout.rfind('}')+1])
if x.get('verdict') in {'VALIDATED','UNVERIFIED','CONFLICT','REJECT'}:return x
except Exception:pass
return {'error':'invalid_fact_checker_json'}
except Exception as e:return {'error':str(e)}
def resolve_fact_checks(c,limit=5):
rows=c.execute("select a.*,s.claim,s.subject,s.validation_status,s.session_key from agent_reviews a join semantic_items s on a.item_key=s.session_key||':'||a.claim_id where a.final_action='PENDING_FACT_CHECK' and a.fact_check_status='PENDING' limit ?",(limit,)).fetchall();out=[]
for r in rows:
item=dict(r);d=run_faktenpruefer(item);ver=d.get('verdict');
if 'error' in d:status='ERROR';final='PENDING_FACT_CHECK'
elif ver=='VALIDATED':status='VALIDATED';final=r['decision'] if r['decision'] in ('CREATE','CONFIRM') else 'REJECT'
elif ver=='CONFLICT':status='CONFLICT';final='CONFLICT'
else:status=ver or 'UNVERIFIED';final='REJECT' if (ver=='REJECT' or (ver=='UNVERIFIED' and r['decision']=='REJECT')) else 'PENDING_FACT_CHECK'
c.execute('update agent_reviews set fact_check_status=?,final_action=?,decision_reason=? where item_key=?',(status,final,(r['decision_reason'] or '')+' | Faktenprüfer: '+str(d.get('reason')),r['item_key']))
c.execute('update semantic_items set knowledge_action=? where session_key=? and claim_id=?',(final,r['session_key'],r['claim_id']));c.commit();out.append({'semantic_item_id':r['item_key'],'verdict':ver,'fact_check_status':status,'final_action':final,'reason':d.get('reason'),'error':d.get('error')})
return out
def escalate_review_required(c,limit=5):
rows=c.execute("select * from semantic_items where knowledge_action='REVIEW_REQUIRED' and (session_key||':'||claim_id) not in (select item_key from agent_reviews) order by source_timestamp limit ?",(limit,)).fetchall();done=[]
for r in rows:
item=dict(r); item['session_key']=r['session_key']; d=run_wissenspruefer(item); key=f"{r['session_key']}:{r['claim_id']}"; decision=d.get('decision') if 'error' not in d else None; final={'CREATE':'CREATE','CONFIRM':'CONFIRM','CONFLICT':'CONFLICT','REJECT':'REJECT'}.get(decision,'ERROR')
target=None
if d.get('needs_fact_check'):final='PENDING_FACT_CHECK'
elif final=='CREATE':
item['statement']=r['claim']; item['source_type']='session_user_instruction'; action,target_path=append_note(note_path(r['project']),r['project'],item,{'session_id':r['source_session_id'],'profile':r['source_profile']}); target=str(target_path); final=action
c.execute('update semantic_items set knowledge_action=?,target=? where session_key=? and claim_id=?',(final,target,r['session_key'],r['claim_id']))
c.execute('insert or ignore into agent_reviews values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)',(key,key,r['claim_id'],r['project'],r['source_session_id'],r['source_profile'],'wissenspruefer','wissenspruefer/1.0.0',time.time(),decision,d.get('reason'),d.get('confidence'),1 if d.get('needs_fact_check') else 0,'PENDING' if d.get('needs_fact_check') else None,final,d.get('error')));c.commit();done.append({'semantic_item_id':key,'decision':decision,'final_action':final,'reason':d.get('reason'),'error':d.get('error'),'target':target})
return done
def extract(project,sid,title,text):
if project!='MAS Visual Library':
return []
# These are explicit stable project facts, validated against the live library DB below.
items=[
claim('masvl.root','Die MAS Visual Library liegt produktiv unter /opt/struktur/mas-visual-library/.',0.99),
claim('masvl.db','Die produktive Datenbank ist /opt/struktur/mas-visual-library/database/visual_library.db.',0.99),
claim('masvl.originals_immutable','Originaldateien werden nicht überschrieben oder gelöscht; visual_id und SHA256 bleiben unverändert.',0.98),
claim('masvl.import_analysis_split','Technischer Import/Katalogisierung und fachliche Bildanalyse durch den Visual-Kurator sind getrennte Prozessschritte.',0.95),
claim('masvl.duplicate_rule','SHA256 dient ausschließlich zur Erkennung exakter Dubletten, nicht als Variantenkriterium.',0.98),
claim('masvl.variant_groups','Asset-Groups modellieren Variantenbeziehungen ausschließlich als Metadatenrelationen.',0.96),
claim('masvl.candidate_lifecycle','Variant-Candidates besitzen einen nachvollziehbaren Lebenszyklus mit OPEN, RESOLVED_GROUPED, RESOLVED_FALSE_POSITIVE und RESOLVED_ALREADY_GROUPED.',0.95),
]
try:
p=Path('/opt/struktur/mas-visual-library/database/visual_library.db'); c=sqlite3.connect('file:'+str(p)+'?mode=ro',uri=True)
visuals=c.execute('select count(*) from visuals').fetchone()[0]; analyses=c.execute('select count(*) from analyses').fetchone()[0]; groups=c.execute('select count(*) from asset_groups').fetchone()[0]; grouped=c.execute('select count(*) from visuals where asset_group_id is not null').fetchone()[0]; candidates=c.execute("select count(*) from variant_candidates where status='OPEN'").fetchone()[0]
families=visuals-(grouped-groups)
items += [claim('masvl.live_inventory',f'Live-Stand: {visuals} Visuals und {analyses} Kuratoranalysen; {groups} Asset-Groups; {grouped} gruppierte und {visuals-grouped} ungruppierte Visuals; {families} Motivfamilien; {candidates} offene Variant-Candidates.',0.99,'session+live_db_validation')]
c.close()
except Exception:
pass
return items
def note_path(project):
safe=re.sub(r'[^A-Za-z0-9ÄÖÜäöüß -]+','',project).strip().replace(' ','-')
return VAULT/'03-Projekte'/safe/'SESSION-KNOWLEDGE.md'
def append_note(path,project,item,source):
path.parent.mkdir(parents=True,exist_ok=True)
marker=f"claim_id: {item['claim_id']}"
now=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
if path.exists():
old=path.read_text(encoding='utf-8',errors='replace')
if marker in old:
if item['statement'] in old:
# Existing identical claim is confirmed; no duplicate note.
return 'CONFIRM',path
# Same stable claim id but different assertion: preserve both and flag it.
now=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
conflict=(f"\n## CONFLICT_REVIEW {item['claim_id']}\n"
f"- existing claim marker: {marker}\n"
f"- new assertion: {item['statement']}\n"
f"- source_session_id: {source['session_id']}\n"
f"- timestamp: {now}\n"
f"- recommendation: human review; do not overwrite existing claim automatically\n")
path.write_text(old+conflict,encoding='utf-8')
return 'CONFLICT',path
prefix=''
else:
prefix=f"---\ntitle: {project} – Session Knowledge\ntype: durable_knowledge\nstatus: active\n---\n\n"
old=''
block=f"\n## {item['claim_id']}\n- {item['statement']}\n- {marker}\n- source_session_id: {source['session_id']}\n- project: {project}\n- agent/profile: {source['profile']}\n- timestamp: {now}\n- source_type: {item['source_type']}\n- confidence: {item['confidence']:.2f}\n- knowledge_action: CREATE\n"
path.write_text(prefix+old+block,encoding='utf-8')
return 'CREATE',path
def review_one(c,profile,dbpath,row):
sid=row['id']; key=f'{profile}:{sid}'
existing=c.execute('select status from session_reviews where session_key=?',(key,)).fetchone()
if existing and existing[0]=='COMPLETED': return semantic_one(c,profile,dbpath,row)
msgs=sqlite3.connect('file:'+dbpath+'?mode=ro',uri=True).execute("select role,content from messages where session_id=? and role='user' order by id",(sid,)).fetchall(); text='\n'.join((x[1] or '') for x in msgs)
project=project_for(row['title'] or '',text); items=extract(project,sid,row['title'] or '',text); source={'session_id':sid,'profile':profile,'source_db':dbpath}
result={'session_id':sid,'project':project,'items':[],'status':'COMPLETED','error':None}
try:
if not VAULT.exists(): raise RuntimeError('OBSIDIAN_VAULT_UNAVAILABLE')
for item in items:
action,path=append_note(note_path(project),project,item,source);result['items'].append({**item,'knowledge_action':action,'target':str(path)})
st='COMPLETED'
except Exception as e:
st='KNOWLEDGE_REVIEW_PENDING';result['status']=st;result['error']=str(e)
values=(key,sid,profile,dbpath,project,st,len(items),int(time.time()),time.time() if st=='COMPLETED' else None,result['error'],VERSION)
if existing:
c.execute('update session_reviews set session_id=?,profile=?,source_db=?,project=?,status=?,knowledge_items_found=?,created=?,completed_at=?,error=?,version=? where session_key=?',values[1:]+(key,))
else:
c.execute('insert into session_reviews values(?,?,?,?,?,?,?,?,?,?,?)',values)
c.commit()
AUDIT.parent.mkdir(parents=True,exist_ok=True);AUDIT.open('a',encoding='utf-8').write(json.dumps({**result,'timestamp':time.time()},ensure_ascii=False)+'\n')
sem=semantic_one(c,profile,dbpath,row);return {'deterministic':result,'semantic':sem}
def main():
ap=argparse.ArgumentParser();ap.add_argument('--once',action='store_true');ap.add_argument('--session-id');ap.add_argument('--agent-limit',type=int,default=5);a=ap.parse_args();c=db();out=[]
paths=[p for p in ROOT.glob('profiles/*/state.db') if p.is_file()]+([ROOT/'state.db'] if (ROOT/'state.db').exists() else [])
for p in paths:
profile=p.parent.name if p.parent.name!='.hermes' else 'default'
try:
q=sqlite3.connect('file:'+str(p)+'?mode=ro',uri=True);q.row_factory=sqlite3.Row
sql='select * from sessions where ended_at is not null and message_count>0'
params=()
if a.session_id: sql+=' and id=?';params=(a.session_id,)
for row in q.execute(sql,params):
r=review_one(c,profile,str(p),row)
if r:out.append(r)
except Exception as e: print('SCAN_ERROR',p,e)
agent=escalate_review_required(c,max(0,a.agent_limit));facts=resolve_fact_checks(c,max(0,a.agent_limit));print(json.dumps({'version':VERSION,'processed':len(out),'agent_escalations':len(agent),'agent_results':agent,'fact_checks':facts,'results':out},ensure_ascii=False))
return 0
if __name__=='__main__':raise SystemExit(main())