diff --git a/README.md b/README.md index de78c22..84e4e78 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,9 @@ Generic case construction and its input contracts are documented in [docs/case-builder-v1.md](docs/case-builder-v1.md). Vcore imports for wowdiag/wowsrc are documented in [docs/vcore-adapters-v1.md](docs/vcore-adapters-v1.md). +Authenticated submission of new cases and completed-case feedback is documented in +[docs/intake-api.md](docs/intake-api.md). The HTTP API creates and queues packages; +external callers do not need to implement the internal package format. Requires Python 3.11 or newer. diff --git a/config/api.example.toml b/config/api.example.toml new file mode 100644 index 0000000..62c433c --- /dev/null +++ b/config/api.example.toml @@ -0,0 +1,20 @@ +[api] +workspace = "../service-data/service" +host = "127.0.0.1" +port = 8080 + +[[projects]] +id = "viano-core" +type = "wow" +profile = "vcore" +adapter = "vcore" +# Trusted server-side analysis settings; never accepted from upload metadata. +# [projects.settings] +# project_config = "project.local.toml" +# llm_config = "llm.early-review.local.toml" +# investigation_config = "investigation.bounded.toml" + +[[clients]] +id = "wow-capture-worker" +token_env = "VBA_INTAKE_TOKEN" +projects = ["viano-core"] diff --git a/docs/intake-api.md b/docs/intake-api.md new file mode 100644 index 0000000..17bdd05 --- /dev/null +++ b/docs/intake-api.md @@ -0,0 +1,168 @@ +# Übergabe-API v1 + +Die API erstellt interne Case-Pakete und veröffentlicht sie atomar in der Inbox. +Externe Programme liefern Metadaten und Dateien; sie erzeugen und bearbeiten keine +internen Manifeste. Die API führt keine Analyse und keine LLM-Aufrufe aus. Ein +dauerhaft laufender Analyseworker und die allgemeine Weboberfläche sind folgende +Ausbauschritte. Bis dahin übernimmt der vorhandene Inbox-Prozessor die Aufnahme. + +## Lokaler Start + +```powershell +.\.venv\Scripts\python.exe -m pip install -e '.[api]' +Copy-Item config/api.example.toml config/api.local.toml +$env:VBA_INTAKE_TOKEN = [Convert]::ToHexString([System.Security.Cryptography.RandomNumberGenerator]::GetBytes(32)) +.\.venv\Scripts\python.exe -m viano_bug_analysis.api check --config config/api.local.toml +.\.venv\Scripts\python.exe -m viano_bug_analysis.api serve --config config/api.local.toml +``` + +Der API-Token ist ein eigener Zugangsschlüssel, kein Gemini-Key. Programme erhalten +je einen Client und einen Token aus einer serverseitigen Umgebungsvariable. Der +Token benötigt 32–256 Zeichen aus `A-Z a-z 0-9 _ -`. Auf älterem Windows PowerShell +kann ein solcher Token über einen Passwortmanager erzeugt und als Variable gesetzt +werden. Der Beispielstart bindet nur an `127.0.0.1:8080`. Für entfernte Nutzung sind +HTTPS und ein passend konfigurierter Reverseproxy erforderlich. Tokens nicht in +URLs oder Git-Konfiguration ablegen. Zugriff auf das Workspace-Dateisystem bleibt +auf den Dienst und vertrauenswürdige Betreiber beschränkt. + +`projects` ist die serverseitige Freigabeliste. Jeder Auftrag benennt `project_id` +und `profile_id`. Er kann keine Repository-URLs, DB-Zugänge, Budgets oder +Konfigurationspfade überschreiben. Optionale `projects.settings` halten die spätere +Worker-Konfiguration bereit; die Eingangsschicht verwendet diese Dateien nicht für +Analysen. Aktuell ist der Importadapter `vcore` für `wow/vcore` implementiert. +Weitere Projekte benötigen passende Adapter/Profile, keine separat gepflegten Forks. + +## Endpunkte + +| Methode/Pfad | Bedeutung | +|---|---| +| `GET /healthz` | Minimale öffentliche Erreichbarkeitsprüfung | +| `GET /v1/profiles` | Für diesen Client freigegebene Projekte/Profile | +| `GET /v1/schema` | Öffentliches JSON-Schema der Übergabemetadaten | +| `POST /v1/submissions` | Multipart-Upload, `201` bei neuer Einreihung, `200` bei Wiederholung | +| `GET /v1/submissions/{id}` | Quittung und aktueller verwalteter Zustand | +| `GET /v1/submissions/{id}/package` | Abgeschlossene Akte herunterladen; sonst `409` | + +Alle Endpunkte außer Health benötigen `Authorization: Bearer TOKEN`. +POST benötigt zusätzlich `Idempotency-Key: frei-gewaehlte-eindeutige-id` (1–128 +Zeichen, Buchstaben/Ziffern sowie Punkt, Unterstrich, Bindestrich; erstes Zeichen +alphanumerisch). Derselbe Client/Schlüssel mit denselben Metadaten und Dateiinhalten +liefert dieselbe Quittung, auch nach einem Neustart. Abweichender Inhalt ergibt +`409 API_IDEMPOTENCY_CONFLICT`. Dateinamen und Multipart-Grenzen zählen nicht als +Inhaltsänderung; Rollen und Reihenfolge der Dateien zählen. Ein zweiter neuer +Auftrag für dasselbe Projekt und dieselbe Case-ID wird zurückgewiesen. + +## Neuer Auftrag + +`metadata` ist ein UTF-8-JSON-Textfeld, kein Datei-Upload. Dateien: genau einmal +`wowdiag`, optional einmal `wowsrc`, optional mehrfach `attachments`. Für Vcore muss +die Case-ID dem Bug des Erfassungsarchivs entsprechen (`BUG-000020`). Beide +Erfassungsarchive müssen zueinander passen. Beispiel `submission.json`: + +```json +{ + "kind": "new_case", + "project_id": "viano-core", + "profile_id": "vcore", + "case_id": "BUG-000020", + "report": { + "title": "Wolpertinger fehlen", + "observations": ["Nach Questannahme waren im untersuchten Gebiet keine Tiere sichtbar."], + "expected_behavior": ["Fangbare Tiere sind im Gebiet vorhanden."] + }, + "versions": { + "reported_server": "commit-des-gemeldeten-serverstands", + "capture_server": "commit-des-erfassungsstands" + } +} +``` + +```powershell +curl.exe --fail-with-body -H "Authorization: Bearer $env:VBA_INTAKE_TOKEN" -H "Idempotency-Key: bug20-capture-001" -F "metadata==4.18,<5"] [project.optional-dependencies] test = ["pytest>=8,<10"] research = ["PyMySQL>=1.1,<2", "zstandard>=0.23,<1"] +api = ["starlette>=0.52.1,<0.53", "uvicorn>=0.35,<1", "python-multipart>=0.0.32,<1"] +api-test = ["httpx>=0.28,<1", "pytest>=8,<10"] [tool.setuptools.packages.find] where = ["src"] diff --git a/src/viano_bug_analysis/api/__init__.py b/src/viano_bug_analysis/api/__init__.py new file mode 100644 index 0000000..ae4407a --- /dev/null +++ b/src/viano_bug_analysis/api/__init__.py @@ -0,0 +1,6 @@ +"""Authenticated intake boundary; no model calls or analysis worker.""" + +from .config import APIConfig, Client, Project, IntakeLimits, APIError +from .service import IntakeService + +__all__ = ['APIConfig', 'Client', 'Project', 'IntakeLimits', 'APIError', 'IntakeService'] diff --git a/src/viano_bug_analysis/api/__main__.py b/src/viano_bug_analysis/api/__main__.py new file mode 100644 index 0000000..a10ee35 --- /dev/null +++ b/src/viano_bug_analysis/api/__main__.py @@ -0,0 +1,27 @@ +import argparse +import json +from .config import APIConfig, APIError + + +def main(): + parser = argparse.ArgumentParser(description='Authenticated BugAnalysis intake; no analysis worker') + parser.add_argument('command', choices=['check', 'serve']) + parser.add_argument('--config', required=True) + args = parser.parse_args() + try: + config = APIConfig.load(args.config) + config.tokens() + if args.command == 'check': + print(json.dumps({'status': 'OK', 'projects': [p.id for p in config.projects]})) + return 0 + import uvicorn + from .app import create_app + uvicorn.run(create_app(config), host=config.host, port=config.port, access_log=False) + return 0 + except APIError as error: + print(json.dumps({'status': 'ERROR', 'code': error.code})) + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/viano_bug_analysis/api/app.py b/src/viano_bug_analysis/api/app.py new file mode 100644 index 0000000..ef30074 --- /dev/null +++ b/src/viano_bug_analysis/api/app.py @@ -0,0 +1,168 @@ +import asyncio +import functools +import hmac +import shutil +import tempfile +from pathlib import Path + +from starlette.applications import Starlette +from starlette.datastructures import UploadFile +from starlette.exceptions import HTTPException +from starlette.requests import Request +from starlette.responses import FileResponse, JSONResponse +from starlette.routing import Route +from starlette.concurrency import run_in_threadpool +from starlette.formparsers import MultiPartParser, MultiPartException + +from .config import APIError +from .contracts import schema +from .service import IntakeService + + +def create_app(config): + service = IntakeService(config) + tokens = config.tokens() + service.recover() + active = 0 + + def authenticate(request): + header = request.headers.get('authorization', '') + if not header.startswith('Bearer ') or not header.isascii(): + raise APIError('API_UNAUTHORIZED', 401) + for client, token in tokens: + if hmac.compare_digest(header[7:], token): + return client + raise APIError('API_UNAUTHORIZED', 401) + + async def health(request): + return JSONResponse({'status': 'OK'}) + + async def profiles(request): + client = authenticate(request) + return JSONResponse({'projects': [{'project_id': p.id, 'profile_id': p.profile, + 'adapter': p.adapter} for p in config.projects if p.id in client.projects]}) + + async def contract(request): + authenticate(request) + return JSONResponse(schema()) + + async def status(request): + client = authenticate(request) + return JSONResponse(await run_in_threadpool(service.status, request.path_params['submission'], client)) + + async def package(request): + client = authenticate(request) + path = await run_in_threadpool(service.package, request.path_params['submission'], client) + return FileResponse(path, media_type='application/zip', filename='case.zip') + + async def submit(request): + nonlocal active + client = authenticate(request) # never parse an unauthenticated upload + if active >= config.limits.max_inflight_uploads: + raise APIError('API_BUSY', 503) + if request.headers.get('content-type', '').split(';')[0] != 'multipart/form-data': + raise APIError('API_CONTENT_TYPE', 415) + active += 1 + form = None + try: + if shutil.disk_usage(service.root).free < config.limits.min_free_bytes: + raise APIError('API_STORAGE_LOW', 507) + async with asyncio.timeout(config.limits.upload_timeout_seconds): + parser = UploadParser(request.headers, request.stream(), + max_files=config.limits.max_files, max_fields=1, + max_part_size=config.limits.max_metadata_bytes, + file_limit=config.limits.max_file_bytes) + form = await parser.parse() + raw = form.get('metadata') + if not isinstance(raw, str): + raise APIError('API_METADATA_INVALID', 422) + with tempfile.TemporaryDirectory(dir=service.staging) as temporary: + uploads = [] + for i, (role, value) in enumerate(form.multi_items()): + if role == 'metadata': + continue + if not isinstance(value, UploadFile): + raise APIError('API_UPLOAD_ROLES', 422) + path = Path(temporary) / str(i) + size = 0 + with path.open('wb') as output: + while chunk := await value.read(65536): + size += len(chunk) + if size > config.limits.max_file_bytes: + raise APIError('API_FILE_LIMIT', 413) + output.write(chunk) + uploads.append((role, path)) + result, replay = await run_in_threadpool(functools.partial(service.submit, client, + request.headers.get('idempotency-key'), raw.encode('utf-8'), uploads)) + return JSONResponse(result, status_code=200 if replay else 201) + except TimeoutError: + raise APIError('API_UPLOAD_TIMEOUT', 408) from None + except MultiPartException: + raise APIError('API_MULTIPART_INVALID', 400) from None + finally: + if form is not None: + await form.close() + active -= 1 + + async def error(request, exc): + return JSONResponse({'status': 'ERROR', 'code': exc.code}, status_code=exc.status) + + async def http_error(request, exc): + return JSONResponse({'status': 'ERROR', 'code': 'API_BAD_REQUEST'}, status_code=exc.status_code) + + async def storage_error(request, exc): + return JSONResponse({'status': 'ERROR', 'code': 'API_STORAGE_ERROR'}, status_code=503) + + app = Starlette(routes=[Route('/healthz', health), Route('/v1/profiles', profiles), + Route('/v1/schema', contract), Route('/v1/submissions', submit, methods=['POST']), + Route('/v1/submissions/{submission}', status), Route('/v1/submissions/{submission}/package', package)], + exception_handlers={APIError: error, HTTPException: http_error, OSError: storage_error}) + app.state.intake = service + return BodyLimit(app, config.limits.max_body_bytes) + + +class UploadParser(MultiPartParser): + """Bound file parts while parsing and close partial spools on cancellation.""" + def __init__(self, *args, file_limit, **kwargs): + super().__init__(*args, **kwargs) + self.file_limit = file_limit + self.file_bytes = 0 + + def on_part_begin(self): + super().on_part_begin() + self.file_bytes = 0 + + def on_part_data(self, data, start, end): + if self._current_part.file is not None: + self.file_bytes += end - start + if self.file_bytes > self.file_limit: + raise APIError('API_FILE_LIMIT', 413) + super().on_part_data(data, start, end) + + async def parse(self): + try: + return await super().parse() + except BaseException: + for file in self._files_to_close_on_error: + file.close() + raise + + +class BodyLimit: + """Count actual ASGI bytes, also for chunked requests without Content-Length.""" + def __init__(self, app, maximum): + self.app, self.maximum = app, maximum + self.state = app.state + + async def __call__(self, scope, receive, send): + if scope['type'] != 'http': + return await self.app(scope, receive, send) + total = 0 + async def bounded(): + nonlocal total + message = await receive() + total += len(message.get('body', b'')) + if total > self.maximum: + raise APIError('API_BODY_LIMIT', 413) + return message + await self.app(scope, bounded, send) diff --git a/src/viano_bug_analysis/api/config.py b/src/viano_bug_analysis/api/config.py new file mode 100644 index 0000000..eaf2312 --- /dev/null +++ b/src/viano_bug_analysis/api/config.py @@ -0,0 +1,132 @@ +import os +import re +import stat +import tomllib +from dataclasses import dataclass, field +from pathlib import Path + +IDENTIFIER = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,127}') +SETTINGS = {'project_config', 'llm_config', 'investigation_config', 'budget_config', + 'analysis_config', 'diagnosis_config', 'change_request_config', 'correction_config'} + + +class APIError(Exception): + def __init__(self, code, status=400): + self.code, self.status = code, status + super().__init__(code) + + +def directory(path): + """Service-owned paths must not pass through links/junctions.""" + path = Path(path).absolute() + for part in [*reversed(path.parents), path]: + if part.exists() or part.is_symlink(): + info = part.lstat() + if stat.S_ISLNK(info.st_mode) or getattr(info, 'st_file_attributes', 0) & 0x400: + raise APIError('API_UNSAFE_STORAGE', 500) + path.mkdir(parents=True, exist_ok=True) + return path.resolve() + + +@dataclass(frozen=True) +class IntakeLimits: + # Operational safety defaults, not permanent package-format constraints. + max_body_bytes: int = 768 * 1024**2 + max_file_bytes: int = 512 * 1024**2 + max_metadata_bytes: int = 64 * 1024 + max_files: int = 12 + upload_timeout_seconds: int = 120 + max_inflight_uploads: int = 2 + min_free_bytes: int = 512 * 1024**2 + + def __post_init__(self): + if any(type(v) is not int or v <= 0 for v in vars(self).values()): + raise APIError('API_CONFIG', 500) + + +@dataclass(frozen=True) +class Project: + id: str + type: str + profile: str + adapter: str = 'vcore' + settings: dict = field(default_factory=dict) + + def __post_init__(self): + if (any(not isinstance(v, str) or not IDENTIFIER.fullmatch(v) + for v in (self.id, self.type, self.profile)) + or self.adapter != 'vcore' or (self.type, self.profile) != ('wow', 'vcore') + or set(self.settings) - SETTINGS): + raise APIError('API_PROFILE_CONFIG', 500) + + +@dataclass(frozen=True) +class Client: + id: str + token_env: str + projects: tuple[str, ...] + + +@dataclass(frozen=True) +class APIConfig: + workspace: Path + projects: tuple[Project, ...] + clients: tuple[Client, ...] + host: str = '127.0.0.1' + port: int = 8080 + limits: IntakeLimits = field(default_factory=IntakeLimits) + + def __post_init__(self): + ids = {p.id for p in self.projects} + if (not ids or len(ids) != len(self.projects) or not self.clients + or len({c.id for c in self.clients}) != len(self.clients) + or type(self.port) is not int or not 0 < self.port < 65536 + or not isinstance(self.host, str) or not self.host): + raise APIError('API_CONFIG', 500) + for c in self.clients: + if (not isinstance(c.id, str) or not IDENTIFIER.fullmatch(c.id) + or not isinstance(c.token_env, str) + or not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', c.token_env) + or not c.projects or not set(c.projects) <= ids): + raise APIError('API_CONFIG', 500) + + def tokens(self): + result = [] + for client in self.clients: + token = os.environ.get(client.token_env, '') + if not re.fullmatch(r'[A-Za-z0-9_-]{32,256}', token): + raise APIError('API_CREDENTIALS', 500) + if any(token == existing for _, existing in result): + raise APIError('API_CREDENTIALS', 500) + result.append((client, token)) + return result + + @classmethod + def load(cls, path): + try: + path = Path(path).resolve() + data = tomllib.loads(path.read_text(encoding='utf-8')) + if set(data) - {'api', 'projects', 'clients', 'limits'}: + raise ValueError() + api = data['api'] + if set(api) - {'workspace', 'host', 'port'}: + raise ValueError() + projects = [] + for row in data['projects']: + row = dict(row) + settings = row.pop('settings', {}) + if set(settings) - SETTINGS: + raise ValueError() + resolved = {} + for key, value in settings.items(): + target = (path.parent / value).resolve() + if not target.is_file(): + raise ValueError() + resolved[key] = target + projects.append(Project(**row, settings=resolved)) + clients = [Client(**(dict(c) | {'projects': tuple(c['projects'])})) for c in data['clients']] + return cls(workspace=(path.parent / api['workspace']).resolve(), + projects=tuple(projects), clients=tuple(clients), host=api.get('host', '127.0.0.1'), + port=api.get('port', 8080), limits=IntakeLimits(**data.get('limits', {}))) + except (OSError, ValueError, TypeError, KeyError): + raise APIError('API_CONFIG', 500) from None diff --git a/src/viano_bug_analysis/api/contracts.py b/src/viano_bug_analysis/api/contracts.py new file mode 100644 index 0000000..3abc42d --- /dev/null +++ b/src/viano_bug_analysis/api/contracts.py @@ -0,0 +1,23 @@ +"""Public metadata contract; clients never describe internal ZIP members.""" +from importlib.resources import files +import json + +from jsonschema import Draft202012Validator +from viano_bug_analysis.package.manifest import parse_manifest +from .config import APIError + + +def schema(): + return json.loads(files('viano_bug_analysis.schemas').joinpath('api-submission-v1.schema.json').read_text(encoding='utf-8')) + + +def metadata(raw, limit): + try: + if len(raw) > limit: + raise APIError('API_METADATA_LIMIT', 413) + result = parse_manifest(raw) + if not Draft202012Validator(schema()).is_valid(result): + raise APIError('API_METADATA_INVALID', 422) + return result + except (ValueError, TypeError, UnicodeError, RecursionError): + raise APIError('API_METADATA_INVALID', 422) from None diff --git a/src/viano_bug_analysis/api/packages.py b/src/viano_bug_analysis/api/packages.py new file mode 100644 index 0000000..679dfbb --- /dev/null +++ b/src/viano_bug_analysis/api/packages.py @@ -0,0 +1,105 @@ +"""Trusted conversion of public submissions into append-only case artifacts.""" +import copy +from pathlib import Path + +from viano_bug_analysis.lifecycle.checkpoints import CheckpointStore +from viano_bug_analysis.lifecycle.job import timestamp +from viano_bug_analysis.lifecycle.storage import Workspace, json_bytes +from viano_bug_analysis.package import PackageReader +from viano_bug_analysis.package.factory import PackageFactory +from viano_bug_analysis.profiles.vcore.importer import VcoreImporter +from viano_bug_analysis.case.contracts import validate +from .config import APIError + +TERMINAL = {'OUTPUT_READY', 'HANDED_OFF', 'FAILED_TO_ANALYZE'} + + +def documents(data, client, submission, uploads, index): + prefix = f'intake/submissions/{submission}' + request_ref = prefix + '/request.json' + additions = {request_ref: json_bytes({'format': 'viano-api-submission', 'format_version': 1, + 'received_at': timestamp(), 'actor': client.id, 'metadata': data})} + report = {'format': 'viano-bug-report', 'format_version': 1} + supplied = data.get('report', {}) + if 'title' in supplied: + report['title'] = supplied['title'] + if 'description' in supplied: + report['summary'] = supplied['description'] + for source, target in [('observations', 'observations'), ('expected_behavior', 'expected_behavior'), + ('tester_assumptions', 'tester_hypotheses'), ('reproduction_steps', 'comments')]: + report[target] = [{'text': text, 'author': client.id, + 'origin': {'source_ref': request_ref, 'json_pointer': f'/metadata/report/{source}/{i}'}} + for i, text in enumerate(supplied.get(source, []))] + if feedback := data.get('feedback'): + report['comments'].append({'text': f"{feedback['outcome']}: {feedback['summary']}", 'author': client.id}) + for field in ('patch_id', 'tested_build'): + if field in feedback: + report['comments'].append({'text': f"Client-supplied {field}: {feedback[field]}", 'author': client.id}) + report['observations'].extend({'text': text, 'author': client.id, + 'origin': {'source_ref': request_ref, 'json_pointer': f'/metadata/feedback/observations/{i}'}} + for i, text in enumerate(feedback['observations'])) + if data.get('versions'): + report['comments'].append({'text': 'Client-supplied, unverified versions: ' + + json_bytes(data['versions']).decode(), 'author': client.id}) + report_ref = prefix + '/report.json' + validate(report, 'report') + additions[report_ref] = json_bytes(report) + index = copy.deepcopy(index) + index.setdefault('reports', []).append(report_ref) + index.setdefault('attachments', []).append(request_ref) + for i, (role, path) in enumerate(uploads): + if role == 'attachments': + ref = prefix + f'/attachments/{i:04d}.bin' + additions[ref] = Path(path).read_bytes() + index['attachments'].append(ref) + index_ref = prefix + '/case-inputs.json' + validate(index, 'case-inputs') + additions[index_ref] = json_bytes(index) + return additions, {'case_inputs': index_ref, 'intake_request': request_ref} + + +def build_new(destination, data, client, project, submission, package_id, uploads): + roles = dict(uploads) + plan = VcoreImporter().prepare(roles['wowdiag'], roles.get('wowsrc')) + # Do not associate a collector archive with a different externally named bug. + if data['case_id'] != f'BUG-{plan.bug_id:06d}': + raise APIError('API_CASE_ID_MISMATCH', 422) + import json + additions, pointers = documents(data, client, submission, uploads, + json.loads(plan.additions[plan.index_ref])) + now = timestamp() + manifest = {'format': 'viano-bug-case-package', 'format_version': 1, 'package_id': package_id, + 'package_generation': 0, 'case_id': data['case_id'], + 'project': {'id': project.id, 'type': project.type, 'profile': project.profile}, + 'created_at': now, 'updated_at': now, 'state': 'NEW', 'analysis_revision': 0, + 'routing': {'destination': 'inbox', 'reason': None, 'retry_count': 0, 'retry_not_before': None}, + 'artifacts': {'case': None, 'diagnosis': None, 'change_request': None, + 'vcore_import': plan.receipt_ref, **pointers}, + 'history': {'analysis_revisions': 0, 'reproductions': plan.reproductions_count, + 'evidence': 0, 'handoffs': 0, 'feedback': 0, 'events_ref': None}, 'integrity': {}} + PackageFactory().create(destination, manifest, plan.additions | additions) + + +def build_return(destination, scratch, data, client, project, submission, uploads): + source = dict(uploads)['case_package'] + with PackageReader(source) as reader: + original = reader.manifest + if (original['case_id'] != data['case_id'] or original['project'] != + {'id': project.id, 'type': project.type, 'profile': project.profile}): + raise APIError('API_CASE_IDENTITY', 422) + if original['state'] not in TERMINAL: + raise APIError('API_CASE_NOT_COMPLETED', 409) + import json + index = json.loads(reader.read_bytes(original['artifacts']['case_inputs'])) + validate(index, 'case-inputs') + additions, pointers = documents(data, client, submission, uploads, index) + store = CheckpointStore(Workspace(scratch)) + store.intake(source) + case_id, package_id = original['case_id'], original['package_id'] + if original['state'] == 'OUTPUT_READY': + store.advance(case_id, package_id, 'HANDED_OFF', step_id=f'api:{submission}:handoff') + result = store.advance(case_id, package_id, 'REOPENED', step_id=f'api:{submission}:reopen', + details={'operation': 'api-case-return', 'submission_id': submission}, additions=additions, + artifacts=pointers, feedback_increment=1) + import shutil + shutil.copyfile(result, destination) diff --git a/src/viano_bug_analysis/api/service.py b/src/viano_bug_analysis/api/service.py new file mode 100644 index 0000000..1d6f20f --- /dev/null +++ b/src/viano_bug_analysis/api/service.py @@ -0,0 +1,193 @@ +"""Durable, idempotent inbox publication. This boundary never invokes an LLM.""" +import hashlib +import json +import os +import shutil +import tempfile +from pathlib import Path + +from viano_bug_analysis.lifecycle.checkpoints import CheckpointStore, checkpoint_chain, fingerprint +from viano_bug_analysis.lifecycle.errors import LifecycleError, LockBusy +from viano_bug_analysis.lifecycle.locking import FileLock +from viano_bug_analysis.lifecycle.storage import Workspace, atomic_json, json_bytes, key +from viano_bug_analysis.package import PackageReader, PackageReadError, PackageWriteError +from viano_bug_analysis.profiles.vcore.errors import AdapterError +from viano_bug_analysis.case.errors import CaseBuildError +from .config import APIError, IDENTIFIER, directory +from .contracts import metadata +from .packages import TERMINAL, build_new, build_return + + +def digest(path): + with open(path, 'rb') as stream: + return hashlib.file_digest(stream, 'sha256').hexdigest() + + +class IntakeService: + def __init__(self, config): + self.config = config + directory(config.workspace) + self.workspace = Workspace(config.workspace) + for name in ('inbox', 'cases', 'runtime', 'runtime/locks', 'runtime/intake'): + directory(self.workspace.root / name) + self.root = self.workspace.runtime / 'intake' + self.requests = directory(self.root / 'requests') + self.staging = directory(self.root / 'staging') + self.store = CheckpointStore(self.workspace) + + def project(self, client, data): + if data['project_id'] not in client.projects: + raise APIError('API_PROJECT_FORBIDDEN', 403) + project = next(p for p in self.config.projects if p.id == data['project_id']) + if project.profile != data['profile_id']: + raise APIError('API_PROFILE_MISMATCH', 422) + return project + + def _receipt(self, submission, client): + if not IDENTIFIER.fullmatch(submission): + raise APIError('API_NOT_FOUND', 404) + path = self.requests / submission / 'receipt.json' + if not path.is_file(): + raise APIError('API_NOT_FOUND', 404) + receipt = json.loads(path.read_bytes()) + if receipt['project_id'] not in client.projects: + raise APIError('API_NOT_FOUND', 404) + return receipt + + def status(self, submission, client): + receipt = self._receipt(submission, client) + status = receipt['status'] + latest = self.store.latest(receipt['case_id'], receipt['package_id']) + if self._known(receipt, latest): + with PackageReader(latest) as reader: + status = reader.manifest['state'] + return {k: receipt[k] for k in ('submission_id', 'project_id', 'profile_id', 'case_id', + 'package_id', 'analysis_revision')} | {'status': status} + + def package(self, submission, client): + receipt = self._receipt(submission, client) + latest = self.store.latest(receipt['case_id'], receipt['package_id']) + if self._known(receipt, latest): + with PackageReader(latest) as reader: + if reader.manifest['state'] in TERMINAL: + return latest + raise APIError('API_RESULT_NOT_READY', 409) + + def _known(self, receipt, head): + if not head: + return False + with PackageReader(head) as reader: + return (fingerprint(reader) == receipt['output_fingerprint'] or + any(c['source_fingerprint'] == receipt['output_fingerprint'] + for _, c in checkpoint_chain(reader))) + + def _publish(self, receipt): + folder = self.requests / receipt['submission_id'] + inbox = self.workspace.inbox / (receipt['submission_id'] + '.zip') + with self.store._locks(receipt['case_id']): + head = self.store.latest(receipt['case_id'], receipt['package_id']) + if inbox.exists(): + if digest(inbox) != receipt['output_sha256']: + raise APIError('API_QUEUE_CONFLICT', 409) + elif not self._known(receipt, head): + actual = None + if head: + with PackageReader(head) as reader: + actual = fingerprint(reader) + if actual != receipt['base_fingerprint']: + raise APIError('API_STALE_CASE', 409) + prepared = folder / 'package.zip' + if not prepared.is_file() or digest(prepared) != receipt['output_sha256']: + raise APIError('API_PREPARED_PACKAGE_MISSING', 500) + # Hard-link publication is atomic and cannot replace a queue entry. + os.link(prepared, inbox) + receipt['status'] = 'QUEUED' + atomic_json(folder / 'receipt.json', receipt) + (folder / 'package.zip').unlink(missing_ok=True) + + def recover(self): + """Resume only committed PREPARED submissions after an interrupted publish.""" + with FileLock(self.workspace.runtime / 'locks/api-intake.lock'): + for path in self.requests.glob('*/receipt.json'): + receipt = json.loads(path.read_bytes()) + if receipt['status'] == 'PREPARED': + self._publish(receipt) + + def submit(self, client, idempotency_key, raw_metadata, uploads): + if not isinstance(idempotency_key, str) or not IDENTIFIER.fullmatch(idempotency_key): + raise APIError('API_IDEMPOTENCY_KEY', 400) + data = metadata(raw_metadata, self.config.limits.max_metadata_bytes) + project = self.project(client, data) + roles = [r for r, _ in uploads] + required = 'wowdiag' if data['kind'] == 'new_case' else 'case_package' + allowed = {'wowdiag', 'wowsrc', 'attachments'} if data['kind'] == 'new_case' else {'case_package', 'attachments'} + if (required not in roles or set(roles) - allowed or len(roles) > self.config.limits.max_files + or any(roles.count(r) > 1 for r in allowed - {'attachments'})): + raise APIError('API_UPLOAD_ROLES', 422) + if any(Path(p).stat().st_size > self.config.limits.max_file_bytes for _, p in uploads): + raise APIError('API_FILE_LIMIT', 413) + submission = key(client.id + ':' + idempotency_key) + request_hash = key(json_bytes({'metadata': data, 'files': [ + [r, Path(p).stat().st_size, digest(p)] for r, p in uploads]}).decode()) + try: + with FileLock(self.workspace.runtime / 'locks/api-intake.lock'): + folder = directory(self.requests / submission) + receipt_path = folder / 'receipt.json' + if receipt_path.exists(): + receipt = json.loads(receipt_path.read_bytes()) + if receipt['request_hash'] != request_hash: + raise APIError('API_IDEMPOTENCY_CONFLICT', 409) + if receipt['status'] == 'PREPARED': + self._publish(receipt) + return self.status(submission, client), True + package_id = 'api-' + key(project.id + ':' + data['case_id']) + base = None + if data['kind'] == 'case_return': + with PackageReader(dict(uploads)['case_package']) as reader: + package_id = reader.manifest['package_id'] + incoming = fingerprint(reader) + head = self.store.latest(data['case_id'], package_id) + if head: + with PackageReader(head) as reader: + base = fingerprint(reader) + if base != incoming: + raise APIError('API_STALE_CASE', 409) + elif self.store.latest(data['case_id'], package_id): + raise APIError('API_CASE_EXISTS', 409) + # An unconsumed submission, including a crash-prepared one, owns this identity. + for path in self.requests.glob('*/receipt.json'): + old = json.loads(path.read_bytes()) + if old['case_id'] == data['case_id'] and old['project_id'] == project.id: + if data['kind'] == 'new_case': + raise APIError('API_CASE_EXISTS', 409) + if old['status'] == 'PREPARED' or (self.workspace.inbox / (old['submission_id'] + '.zip')).exists(): + raise APIError('API_CASE_PENDING', 409) + reserve = sum(Path(p).stat().st_size for _, p in uploads) * 5 + if shutil.disk_usage(self.root).free < self.config.limits.min_free_bytes + reserve: + raise APIError('API_STORAGE_LOW', 507) + with tempfile.TemporaryDirectory(dir=self.staging) as temporary: + destination = Path(temporary) / 'package.zip' + if data['kind'] == 'new_case': + build_new(destination, data, client, project, submission, package_id, uploads) + else: + build_return(destination, Path(temporary) / 'scratch', data, client, project, submission, uploads) + with PackageReader(destination) as reader: + output_fingerprint = fingerprint(reader) + revision = reader.manifest['analysis_revision'] + # Unreferenced package from an interrupted pre-receipt build can be replaced. + os.replace(destination, folder / 'package.zip') + with open(folder / 'package.zip', 'r+b') as stream: + os.fsync(stream.fileno()) + receipt = {'submission_id': submission, 'status': 'PREPARED', 'client_id': client.id, + 'project_id': project.id, 'profile_id': project.profile, 'case_id': data['case_id'], + 'package_id': package_id, 'analysis_revision': revision, 'request_hash': request_hash, + 'base_fingerprint': base, 'output_fingerprint': output_fingerprint, + 'output_sha256': digest(folder / 'package.zip')} + atomic_json(receipt_path, receipt) + self._publish(receipt) + return self.status(submission, client), False + except LockBusy: + raise APIError('API_BUSY', 503) from None + except (PackageReadError, PackageWriteError, AdapterError, LifecycleError, + CaseBuildError, ValueError, TypeError, KeyError, RecursionError): + raise APIError('API_PACKAGE_INVALID', 422) from None diff --git a/src/viano_bug_analysis/lifecycle/checkpoints.py b/src/viano_bug_analysis/lifecycle/checkpoints.py index 25a3ee1..da3c751 100644 --- a/src/viano_bug_analysis/lifecycle/checkpoints.py +++ b/src/viano_bug_analysis/lifecycle/checkpoints.py @@ -283,7 +283,8 @@ class CheckpointStore: reason: str | None = None, retry_not_before: str | None = None, details: dict | None = None, additions: dict[str, bytes] | None = None, artifacts: dict[str, str | None] | None = None, - expected_generation: int | None = None, manual_retry: bool = False) -> Path: + expected_generation: int | None = None, manual_retry: bool = False, + feedback_increment: int = 0) -> Path: try: if not isinstance(step_id, str) or not step_id.strip(): raise LifecycleError("INVALID_STEP", "A nonblank stable step_id is required") @@ -295,11 +296,15 @@ class CheckpointStore: raise LifecycleError("INVALID_GENERATION", "expected_generation must be a nonnegative integer") if type(manual_retry) is not bool: raise LifecycleError("INVALID_RETRY", "manual_retry must be boolean") + if type(feedback_increment) is not int or feedback_increment not in (0, 1) or (feedback_increment and target != State.REOPENED): + raise LifecycleError("INVALID_FEEDBACK", "Feedback may only be appended when reopening") if manual_retry: details = {**(details or {}), "manual_retry": True} payload = {"target": target, "reason": reason, "retry_not_before": retry_not_before, "details": details or {}, "artifacts": artifacts or {}, "additions": {name: hashlib.sha256(data).hexdigest() for name, data in (additions or {}).items()}} + if feedback_increment: + payload['feedback_increment'] = feedback_increment request_hash = hashlib.sha256(json_bytes(payload)).hexdigest() with self._locks(case_id): latest = self.latest(case_id, package_id) @@ -362,6 +367,8 @@ class CheckpointStore: if target == State.REOPENED or (manifest["state"] == State.VALIDATED and target == State.BUILDING_CASE and revision == 0): revision += 1 history = {"analysis_revisions": max(int(manifest["history"]["analysis_revisions"]), revision)} + if feedback_increment: + history['feedback'] = int(manifest['history']['feedback']) + feedback_increment return self._commit(reader, step_id, request_hash, target, details=details, additions=additions, resume_state=resume, updates={"routing": routing, "analysis_revision": revision, "history": history, "artifacts": artifacts or {}}) diff --git a/src/viano_bug_analysis/schemas/api-submission-v1.schema.json b/src/viano_bug_analysis/schemas/api-submission-v1.schema.json new file mode 100644 index 0000000..2770127 --- /dev/null +++ b/src/viano_bug_analysis/schemas/api-submission-v1.schema.json @@ -0,0 +1,31 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", "additionalProperties": false, + "required": ["kind", "project_id", "profile_id", "case_id"], + "properties": { + "kind": {"enum": ["new_case", "case_return"]}, + "project_id": {"$ref": "#/$defs/id"}, "profile_id": {"$ref": "#/$defs/id"}, + "case_id": {"$ref": "#/$defs/id"}, + "report": {"type": "object", "additionalProperties": false, "properties": { + "title": {"$ref": "#/$defs/text"}, "description": {"$ref": "#/$defs/text"}, + "observations": {"$ref": "#/$defs/texts"}, "expected_behavior": {"$ref": "#/$defs/texts"}, + "tester_assumptions": {"$ref": "#/$defs/texts"}, "reproduction_steps": {"$ref": "#/$defs/texts"} + }}, + "versions": {"type": "object", "additionalProperties": false, "properties": { + "reported_server": {"$ref": "#/$defs/text"}, "capture_server": {"$ref": "#/$defs/text"}, + "tested_build": {"$ref": "#/$defs/text"} + }}, + "feedback": {"type": "object", "additionalProperties": false, "required": ["outcome", "summary", "observations"], "properties": { + "outcome": {"enum": ["patch_failed", "validation_failed", "review_feedback", "new_evidence"]}, + "summary": {"$ref": "#/$defs/text"}, "observations": {"$ref": "#/$defs/texts"}, + "patch_id": {"$ref": "#/$defs/text"}, "tested_build": {"$ref": "#/$defs/text"} + }} + }, + "allOf": [{"if": {"properties": {"kind": {"const": "case_return"}}}, + "then": {"required": ["feedback"]}, "else": {"not": {"required": ["feedback"]}}}], + "$defs": { + "id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$"}, + "text": {"type": "string", "minLength": 1, "maxLength": 12000, "pattern": "\\S"}, + "texts": {"type": "array", "minItems": 1, "maxItems": 100, "items": {"$ref": "#/$defs/text"}} + } +} diff --git a/tests/profiles/vcore/test_intake_api.py b/tests/profiles/vcore/test_intake_api.py new file mode 100644 index 0000000..839bda3 --- /dev/null +++ b/tests/profiles/vcore/test_intake_api.py @@ -0,0 +1,212 @@ +import json +from dataclasses import replace + +import pytest +pytest.importorskip('starlette') +pytest.importorskip('httpx') +pytest.importorskip('multipart') +from starlette.testclient import TestClient + +from viano_bug_analysis.api import APIConfig, Client, Project, IntakeLimits +from viano_bug_analysis.api.app import create_app +from viano_bug_analysis.api.config import APIError +from viano_bug_analysis.lifecycle.checkpoints import CheckpointStore +from viano_bug_analysis.lifecycle.storage import Workspace +from viano_bug_analysis.package import PackageReader + +TOKEN = 'test_token_' + 'x' * 40 + + +@pytest.fixture +def api(tmp_path, monkeypatch): + monkeypatch.setenv('TEST_INTAKE_TOKEN', TOKEN) + config = APIConfig(tmp_path / 'service', (Project('core', 'wow', 'vcore'),), + (Client('capture', 'TEST_INTAKE_TOKEN', ('core',)),), limits=IntakeLimits(min_free_bytes=1)) + app = create_app(config) + return TestClient(app), app.state.intake, config + + +def data(kind='new_case'): + result = {'kind': kind, 'project_id': 'core', 'profile_id': 'vcore', 'case_id': 'BUG-000020', + 'report': {'observations': ['The observed bug persists.']}} + if kind == 'case_return': + result['feedback'] = {'outcome': 'patch_failed', 'summary': 'Still broken', + 'observations': ['The tested patch did not fix it.']} + return result + + +def post(client, source, *, metadata=None, key='first', role='wowdiag', token=TOKEN): + return client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + token, + 'Idempotency-Key': key}, data={'metadata': json.dumps(metadata or data())}, + files={role: ('../untrusted.bin', source.read_bytes(), 'application/octet-stream')}) + + +def test_create_replay_conflict_and_private_status(api, source_pair): + client, service, _ = api + source = source_pair[0] + response = post(client, source) + assert response.status_code == 201, response.text + receipt = response.json() + assert receipt['status'] == 'QUEUED' + assert post(client, source).status_code == 200 + assert len(list(service.workspace.inbox.glob('*.zip'))) == 1 + changed = data() | {'report': {'title': 'Changed'}} + assert post(client, source, metadata=changed).status_code == 409 + assert post(client, source, key='different').status_code == 409 + assert post(client, source, token='bad').status_code == 401 + assert client.get('/v1/submissions/' + receipt['submission_id']).status_code == 401 + with PackageReader(next(service.workspace.inbox.glob('*.zip'))) as reader: + assert reader.manifest['project']['id'] == 'core' + assert reader.manifest['state'] == 'NEW' + assert b'The observed bug persists.' in reader.read_bytes( + 'intake/submissions/' + receipt['submission_id'] + '/report.json') + + +def test_return_preserves_history_and_increments_revision_once(api, source_pair, tmp_path): + client, service, _ = api + first = post(client, source_pair[0]).json() + inbox = next(service.workspace.inbox.glob('*.zip')) + # Standalone terminal case simulates transfer back from a different deployment. + from viano_bug_analysis.package import PackageWriter + terminal = tmp_path / 'terminal.zip' + PackageWriter().write(inbox, terminal, manifest_updates={'state': 'OUTPUT_READY', + 'analysis_revision': 1, 'history': {'analysis_revisions': 1}, + 'routing': {'destination': 'outbox'}}) + inbox.unlink() + response = post(client, terminal, metadata=data('case_return'), key='return', role='case_package') + assert response.status_code == 201, response.text + returned = next(service.workspace.inbox.glob('*.zip')) + with PackageReader(terminal) as old, PackageReader(returned) as new: + assert new.manifest['state'] == 'REOPENED' + assert new.manifest['analysis_revision'] == 2 + assert new.manifest['history']['feedback'] == 1 + assert new.manifest['package_id'] == first['package_id'] + for ref in old.files: + if ref != 'manifest.json': + assert old.read_bytes(ref) == new.read_bytes(ref) + index = json.loads(new.read_bytes(new.manifest['artifacts']['case_inputs'])) + assert any(b'The tested patch did not fix it.' in new.read_bytes(ref) for ref in index['reports']) + assert post(client, terminal, metadata=data('case_return'), key='return', role='case_package').status_code == 200 + assert post(client, terminal, metadata=data('case_return'), key='return2', role='case_package').status_code == 409 + from viano_bug_analysis.case import CaseBuilder + isolated = CheckpointStore(Workspace(tmp_path / 'analysis')) + isolated.intake(returned) + built = CaseBuilder(isolated).build('BUG-000020', first['package_id']) + with PackageReader(built.package) as reader: + case = json.loads(reader.read_bytes(built.case_ref)) + assert any(o['text'] == 'The tested patch did not fix it.' + for o in case['human_observations']['observations']) + + +def test_active_case_cannot_be_returned(api, source_pair): + client, service, _ = api + post(client, source_pair[0]) + source = next(service.workspace.inbox.glob('*.zip')) + assert post(client, source, metadata=data('case_return'), key='return', role='case_package').status_code == 409 + + +def test_limits_and_contract(api, source_pair): + client, _, config = api + assert client.get('/healthz').json() == {'status': 'OK'} + assert client.get('/v1/schema', headers={'Authorization': 'Bearer ' + TOKEN}).status_code == 200 + assert post(client, source_pair[0], metadata=data() | {'budget': 99}).status_code == 422 + assert post(client, source_pair[0], metadata=data() | {'project_id': 'elsewhere'}).status_code == 403 + tiny = replace(config, workspace=config.workspace.parent / 'tiny', + limits=replace(config.limits, max_body_bytes=100)) + limited = TestClient(create_app(tiny)) + assert post(limited, source_pair[0]).status_code == 413 + + +def test_recover_prepared_after_publication_interruption(api, source_pair, monkeypatch): + client, service, _ = api + publish = service._publish + def interrupted(receipt): + raise OSError('simulated crash') + monkeypatch.setattr(service, '_publish', interrupted) + with pytest.raises(OSError): + service.submit(service.config.clients[0], 'crash', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) + monkeypatch.setattr(service, '_publish', publish) + service.recover() + assert len(list(service.workspace.inbox.glob('*.zip'))) == 1 + result, replay = service.submit(service.config.clients[0], 'crash', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) + assert replay and result['status'] == 'QUEUED' + + +def test_published_receipt_crash_replay_after_worker_consumes(api, source_pair, monkeypatch): + import viano_bug_analysis.api.service as module + client, service, _ = api + original = module.atomic_json + def interrupted(path, value): + if value.get('status') == 'QUEUED': + raise OSError('crash after publication') + original(path, value) + monkeypatch.setattr(module, 'atomic_json', interrupted) + with pytest.raises(OSError): + service.submit(service.config.clients[0], 'crash2', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) + inbox = next(service.workspace.inbox.glob('*.zip')) + service.store.intake(inbox) + inbox.unlink() + monkeypatch.setattr(module, 'atomic_json', original) + service.recover() + assert not list(service.workspace.inbox.glob('*.zip')) + result, replay = service.submit(service.config.clients[0], 'crash2', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) + assert replay and result['status'] == 'VALIDATED' + + +def test_file_limit_and_partial_upload_cleanup(api, source_pair): + _, _, config = api + tiny = replace(config, workspace=config.workspace.parent / 'files', + limits=replace(config.limits, max_file_bytes=5)) + app = create_app(tiny) + response = post(TestClient(app), source_pair[0]) + assert response.status_code == 413 + assert not list(app.state.intake.staging.iterdir()) + + +def test_return_with_stale_managed_head_is_rejected(api, source_pair, tmp_path): + client, service, _ = api + first = post(client, source_pair[0]).json() + inbox = next(service.workspace.inbox.glob('*.zip')) + from viano_bug_analysis.package import PackageWriter + terminal = tmp_path / 'terminal.zip' + PackageWriter().write(inbox, terminal, manifest_updates={'state': 'OUTPUT_READY', + 'analysis_revision': 1, 'history': {'analysis_revisions': 1}, 'routing': {'destination': 'outbox'}}) + service.store.intake(terminal) + inbox.unlink() + response = post(client, terminal, metadata=data('case_return'), key='stale', role='case_package') + assert response.status_code == 409 + head = service.store.latest('BUG-000020', first['package_id']) + response = post(client, head, metadata=data('case_return'), key='fresh', role='case_package') + assert response.status_code == 201, response.text + assert response.json()['status'] == 'QUEUED' # not the previous OUTPUT_READY result + auth = {'Authorization': 'Bearer ' + TOKEN} + assert client.get('/v1/submissions/' + response.json()['submission_id'] + '/package', headers=auth).status_code == 409 + + +def test_scope_invalid_json_and_mismatched_source(api, source_pair): + client, service, _ = api + response = post(client, source_pair[0], metadata=data() | {'case_id': 'BUG-000021'}) + assert response.status_code == 422 + assert not list(service.workspace.inbox.glob('*.zip')) + raw = client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + TOKEN, + 'Idempotency-Key': 'duplicate-json'}, data={'metadata': '{"kind":"new_case","kind":"case_return"}'}, + files={'wowdiag': ('source', source_pair[0].read_bytes())}) + assert raw.status_code == 422 + first = post(client, source_pair[0]).json() + stranger = Client('stranger', 'OTHER', ('other-project',)) + with pytest.raises(APIError) as exc: + service.status(first['submission_id'], stranger) + assert exc.value.status == 404 + + +def test_pair_and_attachments_are_retained(api, source_pair): + client, service, _ = api + response = client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + TOKEN, + 'Idempotency-Key': 'paired'}, data={'metadata': json.dumps(data())}, files=[ + ('wowdiag', ('diag', source_pair[0].read_bytes())), + ('wowsrc', ('src', source_pair[1].read_bytes())), + ('attachments', ('a', b'opaque-human-notes'))]) + assert response.status_code == 201, response.text + with PackageReader(next(service.workspace.inbox.glob('*.zip'))) as reader: + index = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_inputs'])) + assert any(reader.read_bytes(ref) == b'opaque-human-notes' for ref in index['attachments'])