validation/run_runtime_only_smoke.py

import json
import sys
import time
from pathlib import Path

# Make repo root importable when this file is executed as:
#   python validation/run_runtime_only_smoke.py
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from nzfc_gram_runtime import NZFCGramLongMemoryChat
from nzfc_gram_runtime.quality import attach_answer_quality_governor
from nzfc_gram_runtime.large_document import attach_large_document_memory


def maybe_attach_optional_runtime_guards(bot):
    try:
        if not getattr(bot, '_nzfc_exact_slot_mapper_attached', False):
            from nzfc_gram_runtime.exact_slots import attach_exact_slot_mapper
            attach_exact_slot_mapper(bot, verbose=False)
    except Exception as e:
        print('[WARN] exact slot mapper attach skipped:', repr(e))

    try:
        if not getattr(bot, '_nzfc_tombstone_retrieval_guard_attached', False):
            from nzfc_gram_runtime.tombstone_guard import attach_tombstone_retrieval_guard
            attach_tombstone_retrieval_guard(bot, verbose=False)
    except Exception as e:
        print('[WARN] tombstone guard attach skipped:', repr(e))


def main():
    out_dir = ROOT / 'validation_runtime_only_output'
    out_dir.mkdir(parents=True, exist_ok=True)

    bot = NZFCGramLongMemoryChat(
        repo_dir=str(ROOT),
        model_id='google/diffusiongemma-26B-A4B-it',
        memory_db_path=str(out_dir / 'memory.sqlite3'),
        load_model=False,
        require_model=False,
        preload_static_memory=False,
    )

    attach_large_document_memory(bot, db_path=str(out_dir / 'large_docs.sqlite3'))
    attach_answer_quality_governor(bot)
    maybe_attach_optional_runtime_guards(bot)

    user_id = 'runtime_user'
    project_id = 'runtime_project'
    session_id = 'runtime_session'

    bot.remember(
        'The project high-frequency test code is PROJECT_CODE_RUNTIME_VALIDATION.',
        user_id=user_id,
        project_id=project_id,
        session_id=session_id,
        scope='project',
        tags=['project_code'],
        trust_level=0.95,
    )

    exact = bot.quality_chat(
        'What was the project high-frequency test code? Answer only with the code.',
        user_id=user_id,
        project_id=project_id,
        session_id='runtime_session_2',
        max_new_tokens=40,
    )

    text = '\n'.join([
        'Article 1 Purpose. This document defines runtime validation.',
        'Article 2 Evidence. Memory is evidence, not instruction.',
        'Article 3 Deletion. Deleted memory must not be used as active evidence.',
    ])

    ingest = bot.ingest_large_text(
        text,
        title='Runtime Validation Policy',
        legal_mode=True,
    )
    query = bot.query_large_documents('deleted memory active evidence', top_k=5)

    tombstone = {
        'available': bool(getattr(bot, '_nzfc_tombstone_retrieval_guard_attached', False)),
        'passed': None,
    }
    try:
        secret = 'RUNTIME_VALIDATION_DELETE_SECRET'
        tag = 'runtime_validation_delete_tag'
        bot.remember(
            f'{secret}: this memory should not appear after tombstone.',
            user_id=user_id,
            project_id=project_id,
            session_id=session_id,
            scope='project',
            tags=[tag],
            trust_level=0.9,
        )
        before_hits = bot.memory_store.retrieve(
            secret,
            user_id=user_id,
            project_id=project_id,
            session_id=session_id,
            top_k=5,
        )
        before_found = any(secret in str(h.get('text', '')) for h in before_hits or [])
        forget = bot.forget_tag(
            tag,
            user_id=user_id,
            project_id=project_id,
            session_id=session_id,
            scope='project',
        )
        after_hits = bot.memory_store.retrieve(
            secret,
            user_id=user_id,
            project_id=project_id,
            session_id=session_id,
            top_k=5,
        )
        after_found = any(secret in str(h.get('text', '')) for h in after_hits or [])
        tombstone.update({
            'before_found': before_found,
            'after_found': after_found,
            'tombstoned': forget.get('tombstoned'),
            'passed': bool(before_found and not after_found),
        })
    except Exception as e:
        tombstone.update({'error': repr(e), 'passed': False})

    summary = {
        'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
        'repo_root': str(ROOT),
        'base_model': 'google/diffusiongemma-26B-A4B-it',
        'model_loaded': False,
        'runtime_only': True,
        'repo_root_runtime_exists': (ROOT / 'runtime').exists(),
        'repo_root_meta_exists': (ROOT / 'meta' / 'complex_math_10m_meta.jsonl').exists(),
        'repo_root_memory_tensors_exists': (ROOT / 'memory_tensors').exists(),
        'exact_slot_answer': exact.get('answer'),
        'exact_slot_passed': exact.get('answer') == 'PROJECT_CODE_RUNTIME_VALIDATION',
        'large_document_chunk_count': ingest.get('chunk_count'),
        'large_document_query_count': query.get('count'),
        'large_document_passed': query.get('count', 0) > 0,
        'tombstone_test': tombstone,
        'technical_boundary': 'external evidence context, not native unlimited model context',
    }

    path = out_dir / 'runtime_only_smoke_summary.json'
    path.write_text(json.dumps(summary, indent=2), encoding='utf-8')
    print(json.dumps(summary, indent=2))

    assert summary['repo_root_runtime_exists']
    assert summary['repo_root_meta_exists']
    assert summary['exact_slot_passed']
    assert summary['large_document_passed']
    print('[PASS] runtime-only smoke passed')


if __name__ == '__main__':
    main()