software/src/pipeline/request_archive.py

1"""Private, append-only records of actual CLI attempts (never public by default)."""2import hashlib3import json4from datetime import datetime, timezone5from pathlib import Path6from uuid import uuid4789class RequestArchive:10    def __init__(self, directory):11        self.directory = Path(directory)1213    def start(self, model, system, user):14        self.directory.mkdir(parents=True, exist_ok=True, mode=0o700)15        identifier = uuid4().hex16        self._write(identifier + '.input.json', {17            'schema': 1, 'id': identifier, 'timestamp': self.now(), 'requested_model': model,18            'system_prompt': system, 'user_prompt': user,19            'system_prompt_sha256': self.digest(system), 'user_prompt_sha256': self.digest(user),20            'scope': 'Exact CLI prompt arguments; provider-internal context is not observable.'})21        return identifier2223    def finish(self, identifier, stdout='', stderr='', returncode=None, error=None):24        self._write(identifier + '.output.json', {25            'schema': 1, 'id': identifier, 'timestamp': self.now(), 'stdout': stdout,26            'stderr': stderr, 'returncode': returncode, 'error': error,27            'stdout_sha256': self.digest(stdout), 'stderr_sha256': self.digest(stderr)})2829    def _write(self, name, record):30        # Exclusive creation: retries and concurrent workers never overwrite history.31        with (self.directory / name).open('x', encoding='utf-8') as stream:32            stream.write(json.dumps(record, ensure_ascii=False) + '\n')33        (self.directory / name).chmod(0o600)3435    @staticmethod36    def digest(text):37        return hashlib.sha256(text.encode()).hexdigest()3839    @staticmethod40    def now():41        return datetime.now(timezone.utc).isoformat()42