#!/usr/bin/env python3
"""Explicit advanced live receipt. Protected credentials; bounded disposable data.

Never changes existing memory/settings, runs scheduled tasks, or prints credentials.
Research is one ordinary Hermes run. Deletes only objects created by this script.
"""
import argparse
import asyncio
import json
import os
import secrets
import struct
import time
import zlib
from pathlib import Path

import httpx


def png():
    def chunk(kind, data):
        return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data))
    pixels = b''.join(b'\x00' + b'\xff\x00\x00' * 16 + b'\x00\x00\xff' * 16 for _ in range(16))
    return b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', 32, 16, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(pixels)) + chunk(b'IEND', b'')


def pdf(text):
    objects = [b'<< /Type /Catalog /Pages 2 0 R >>', b'<< /Type /Pages /Kids [3 0 R] /Count 1 >>',
               b'<< /Type /Page /Parent 2 0 R /MediaBox [0 0 300 200] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>',
               b'<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>']
    stream = ('BT /F1 12 Tf 20 100 Td (' + text + ') Tj ET').encode('ascii')
    objects.append(b'<< /Length ' + str(len(stream)).encode() + b' >>\nstream\n' + stream + b'\nendstream')
    output = bytearray(b'%PDF-1.4\n'); offsets = [0]
    for number, obj in enumerate(objects, 1):
        offsets.append(len(output)); output.extend(str(number).encode() + b' 0 obj\n' + obj + b'\nendobj\n')
    xref = len(output)
    output.extend(b'xref\n0 6\n0000000000 65535 f \n')
    for offset in offsets[1:]: output.extend(f'{offset:010d} 00000 n \n'.encode())
    output.extend(f'trailer\n<< /Size 6 /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n'.encode())
    return bytes(output)


async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--url', default='https://192.168.86.2:9178')
    parser.add_argument('--credentials', default='.runtime/deployment.json')
    parser.add_argument('--receipt', default='.runtime/chat-features-live-receipt.json')
    args = parser.parse_args()
    source = Path(args.credentials)
    if source.is_symlink() or source.stat().st_mode & 0o077:
        raise RuntimeError('Credentials must be private regular file')
    credentials = json.loads(source.read_text())
    stamp = secrets.token_hex(6)
    receipt = {'url': args.url, 'started_at': time.time(), 'checks': {}, 'cleanup_pending': []}
    attachments, conversations, tasks = [], [], []
    failure = None
    async with httpx.AsyncClient(base_url=args.url, verify=False, timeout=180, follow_redirects=False) as client:
        async def request(method, path, **kwargs):
            response = await client.request(method, path, **kwargs)
            if response.status_code >= 400:
                raise RuntimeError(f'{method} {path.split("?")[0]} status {response.status_code}')
            return response
        async def upload(data, mime, name):
            item = (await request('POST', '/api/attachments', content=data, headers={'Content-Type': mime, 'X-Filename': name})).json()['attachment']
            attachments.append(item['id']); return item
        async def turn(content, ids=(), mode=None):
            conversation = (await request('POST', '/api/conversations', json={'title': 'Advanced qualification ' + stamp})).json()['conversation']
            conversations.append(conversation['id'])
            body = {'content': content, 'attachment_ids': list(ids), 'idempotency_key': secrets.token_hex(16)}
            if mode: body['mode'] = mode
            accepted = (await request('POST', '/api/conversations/' + conversation['id'] + '/messages', json=body)).json()
            rid = accepted['run']['id']
            async with client.stream('GET', '/api/runs/' + rid + '/events') as response:
                if response.status_code != 200: raise RuntimeError('Run events unavailable')
                async for _ in response.aiter_lines(): pass
            detail = (await request('GET', '/api/runs/' + rid)).json()
            if detail['run']['status'] != 'completed': raise RuntimeError('Native Hermes run did not complete')
            return '\n'.join(message['content'] for message in detail['messages'] if message['role'] == 'assistant' and message['status'] == 'completed')
        try:
            login = (await request('POST', '/api/auth/login', json={'username': 'admin', 'password': credentials['admin_password']})).json()
            client.cookies.clear(); client.headers['Authorization'] = 'Bearer ' + login['token']
            caps = (await request('GET', '/api/capabilities')).json()['features']
            receipt['capabilities'] = {key: caps.get(key) for key in ('attachments', 'pdf_extraction', 'image_input', 'scheduled_tasks', 'memory_controls', 'research', 'connectors')}
            text_code, pdf_code = 'TXT' + stamp, 'PDF' + stamp
            text = await upload(('Qualification token: ' + text_code).encode(), 'text/plain', 'qualification.txt')
            assert text['extraction_status'] == 'ready'
            assert text_code in await turn('Read the attached text and return its qualification token only. Do not use tools.', [text['id']])
            receipt['checks']['text_attachment_native_reply'] = True
            document = await upload(pdf('Qualification token ' + pdf_code), 'application/pdf', 'qualification.pdf')
            assert document['extraction_status'] == 'ready'
            assert pdf_code in await turn('Read the attached document and return its qualification token only. Do not use tools.', [document['id']])
            receipt['checks']['pdf_extraction_native_reply'] = True
            image = await upload(png(), 'image/png', 'qualification.png')
            answer = await turn('Name the two panel colors left to right in this attached image. Answer two colors only. Do not use tools.', [image['id']])
            assert 'red' in answer.lower() and 'blue' in answer.lower() and answer.lower().index('red') < answer.lower().index('blue')
            receipt['checks']['image_native_vision_reply'] = True
            memory = (await request('GET', '/api/memory')).json()
            assert {item['section'] for item in memory['sections']} == {'memory', 'user'}
            assert all(len(item['revision']) == 64 for item in memory['sections'])
            receipt['checks']['memory_read_only'] = True
            del memory  # Never include existing private memory content in receipt.
            task = (await request('POST', '/api/tasks', json={'name': 'Qualification ' + stamp, 'schedule': '2099-01-01T00:00:00Z',
                                                          'prompt': 'Qualification placeholder; do not execute before scheduled time.', 'paused': True, 'repeat': 1})).json()['task']
            tasks.append(task['id'])
            for action in ('resume', 'pause'):
                await request('POST', '/api/tasks/' + task['id'] + '/' + action)
            await request('PATCH', '/api/tasks/' + task['id'], json={'name': 'Qualification edited ' + stamp, 'enabled': False})
            listed = (await request('GET', '/api/tasks')).json()['tasks']
            assert any(item['id'] == task['id'] for item in listed)
            await request('DELETE', '/api/tasks/' + task['id']); tasks.remove(task['id'])
            receipt['checks']['task_create_pause_resume_edit_delete_no_run'] = True
            research = (await request('GET', '/api/research')).json()
            assert research['available'] and research['mode'] == 'hermes_tools'
            answer = await turn('Use web research only to verify what organization publishes the Python programming language official documentation. '
                                'Consult https://docs.python.org/ and cite that source URL. Do not write files, run shell commands, schedule jobs or modify memory.', mode='research')
            assert 'https://docs.python.org' in answer
            receipt['checks']['native_research_cited_reply'] = True
        except Exception as error:
            # Never serialize server bodies/provider exception details.
            failure = type(error).__name__ + ': ' + str(error) if isinstance(error, (AssertionError, RuntimeError)) else type(error).__name__
            receipt['failure'] = failure
        finally:
            for kind, ids in (('tasks', tasks), ('attachments', attachments), ('conversations', conversations)):
                for identifier in ids:
                    response = await client.delete('/api/' + kind + '/' + identifier)
                    if response.status_code not in (200, 404):
                        receipt['cleanup_pending'].append({'kind': kind, 'id': identifier, 'status': response.status_code})
            if 'Authorization' in client.headers:
                await client.post('/api/auth/logout')
    receipt['finished_at'] = time.time()
    target = Path(args.receipt); target.parent.mkdir(parents=True, exist_ok=True)
    if target.is_symlink(): raise RuntimeError('Receipt path cannot be a symlink')
    fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    os.fchmod(fd, 0o600)
    with os.fdopen(fd, 'w') as stream: json.dump(receipt, stream, indent=2)
    print('Advanced live checks:', ', '.join(receipt['checks']))
    if failure:
        print('Qualification incomplete:', failure)
        raise SystemExit(1)


if __name__ == '__main__': asyncio.run(main())