From b8c0d7ec40ba242022d233e8ebbee07b7d9490f6 Mon Sep 17 00:00:00 2001 From: Nitsu45 Date: Fri, 23 Oct 2026 13:33:28 +0200 Subject: [PATCH] Add capture worker v0.4 with durable BugAnalysis API handoff --- deploy/capture-worker/README.md | 73 +++ .../buganalysis.config.example.toml | 12 + deploy/capture-worker/buganalysis_handoff.py | 162 +++++ deploy/capture-worker/config.toml.example | 29 + deploy/capture-worker/wowdiag_worker.py | 617 ++++++++++++++++++ .../capture-worker-handoff-20261023.md | 37 ++ tests/profiles/vcore/test_capture_handoff.py | 103 +++ 7 files changed, 1033 insertions(+) create mode 100644 deploy/capture-worker/README.md create mode 100644 deploy/capture-worker/buganalysis.config.example.toml create mode 100644 deploy/capture-worker/buganalysis_handoff.py create mode 100644 deploy/capture-worker/config.toml.example create mode 100644 deploy/capture-worker/wowdiag_worker.py create mode 100644 docs/test-results/capture-worker-handoff-20261023.md create mode 100644 tests/profiles/vcore/test_capture_handoff.py diff --git a/deploy/capture-worker/README.md b/deploy/capture-worker/README.md new file mode 100644 index 0000000..1a9b421 --- /dev/null +++ b/deploy/capture-worker/README.md @@ -0,0 +1,73 @@ +# wowdiag-worker v0.4 – BugAnalysis-Übergabe + +Erweiterung des vorgelegten v0.3-Workers. Die ursprüngliche Queue-Claim-Logik und +DB-Struktur bleiben erhalten. Keine Änderung der aktiven `config.toml`, kein +Dienststart und keine Datenbankmigration durch diese Dateien. + +Nach erfolgreicher wowdiag-/wowsrc-Erzeugung wird ein Versandauftrag dauerhaft +in einer separaten lokalen SQLite-Outbox gespeichert. `diagnostic_runs.completed` +bedeutet weiterhin **Erfassung abgeschlossen**, nicht Analyse abgeschlossen oder +API-Zustellung bestätigt. Die Versandquittung ist über `--handoff-status` sichtbar. +Die API übernimmt den Fall anschließend in ihren eigenen Analyseworker. + +## Einrichtung + +1. Bestehende `wowdiag_worker.py` sichern und die v0.4-Datei sowie + `buganalysis_handoff.py` nebeneinander installieren. Python 3.11 und PyMySQL + bleiben die Voraussetzungen; der HTTP-Client benötigt keine neue Bibliothek. +2. Den Abschnitt aus `buganalysis.config.example.toml` in die vorhandene + Worker-Konfiguration übernehmen. Tatsächliche URL und Projektzuordnung setzen. + Der URL muss die Serverwurzel sein; HTTPS ist für entfernte Server erforderlich. +3. Einen Integrationsschlüssel in BugAnalysis für das Projekt erzeugen und im + Dienst-Environment als `VBA_INTAKE_TOKEN` hinterlegen. Unter Linux eine geschützte + systemd-EnvironmentFile verwenden; den Schlüssel nicht in Skriptargumente, + Repository oder Worker-Logs schreiben. Bestehende Unit um EnvironmentFile + ergänzen und anschließend kontrolliert neu starten. +4. SQLite-Outbox und Diagnosearchive bis zur bestätigten Zustellung erhalten. + Das Outbox-Verzeichnis muss für den Worker schreibbar sein und einen lokalen + Datenträger mit SQLite-Dateisperren verwenden. Die Datei nicht zwischen + unabhängigen Installationen teilen. + +Ohne aktivierten `[buganalysis]`-Abschnitt bleibt der bisherige Ablauf erhalten. +Das Flag bedeutet ausschließlich Erfassung plus Übergabe; der Worker löst keine +Modellanfrage selbst aus. Ein laufender BugAnalysis-Worker kann nach Annahme +allerdings sofort eine kostenpflichtige Analyse beginnen. + +## Wiederholungen und Status + +```sh +python3 wowdiag_worker.py --config config.toml --handoff-status +python3 wowdiag_worker.py --config config.toml --retry-handoffs +``` + +Diese beiden Befehle führen keine Erfassung aus. Der normale Dauerdienst sendet +fällige Aufträge vor dem nächsten Capture; `--once` versucht sie auch danach. +Netzwerkfehler, Serverausfälle, temporäre Ablehnungen und fehlende/ungültige +Zugangsdaten bleiben `pending`. Nach Behebung werden dieselben Inhalte mit +demselben Idempotenzschlüssel versandt. Auch eine verlorene Erfolgsantwort legt +dadurch keinen zweiten Fall an. Es werden keine HTTP-Redirects verfolgt. + +Unveränderliche Metadaten, Ziel-URL, Archivpfade und Inhaltsprüfwerte stehen in der +Outbox; Zugangsdaten stehen dort nicht. Archivänderung oder endgültiger HTTP- +Konflikt wird `attention` und benötigt Betreiberprüfung. Ein Lease verhindert +gleichzeitigen regulären Versand durch mehrere Worker-Prozesse. Ein Prozessabbruch +lässt den Auftrag nach Ablauf des Lease erneut versuchen. Es gibt keine automatische +Neuerzeugung von Archiven oder ein Zurücksetzen der ursprünglichen Queue-Claims. + +Bereits vor dem Update abgeschlossene Erfassungen werden nicht automatisch +nachträglich eingesendet. Ein erneuter neuer Auftrag für dieselbe Bug-ID wird von +BugAnalysis bewusst abgewiesen, wenn der Fall bereits existiert. Rückgaben nach +einem Patch erfolgen über den separaten `case_return`-Vertrag und nicht durch +automatisches Umdeuten neuer Captures. + +Der Übergabeumschlag nennt die Bug-ID und den Erfassungs-Commit. Die wirklichen +Testerbeobachtungen, Titel und Beziehungen liest der BugAnalysis-Importer direkt +aus den unveränderten Archiven. Der Worker erfindet keine Beobachtungen und gibt +den Erfassungsstand nicht als Stand der ursprünglichen Fehlermeldung aus. + +## Lokale Prüfung im BugAnalysis-Checkout + +`python -m pytest tests/profiles/vcore/test_capture_handoff.py` prüft den Versand gegen die echte +lokale Intake-API mit synthetischen Archiven. Kein echter Server, keine Cloud- +Anfrage, keine Änderung der Diagnose-Queue. Ein Server-Feldtest mit einem neuen +Bug und Dienstneustart steht nach der Konfiguration noch aus. diff --git a/deploy/capture-worker/buganalysis.config.example.toml b/deploy/capture-worker/buganalysis.config.example.toml new file mode 100644 index 0000000..0963446 --- /dev/null +++ b/deploy/capture-worker/buganalysis.config.example.toml @@ -0,0 +1,12 @@ +# Add this section to the EXISTING worker configuration. Do not replace its DB/path settings. +[buganalysis] +enabled = true +url = "https://buganalysis.example.org" +project_id = "viano-core" +profile_id = "vcore" +# Keep this name stable and unique for this capture installation / queue database. +producer_id = "vcore-testserver" +token_env = "VBA_INTAKE_TOKEN" +outbox = "buganalysis-outbox.sqlite3" +timeout_seconds = 120 +retry_seconds = 60 diff --git a/deploy/capture-worker/buganalysis_handoff.py b/deploy/capture-worker/buganalysis_handoff.py new file mode 100644 index 0000000..fa37382 --- /dev/null +++ b/deploy/capture-worker/buganalysis_handoff.py @@ -0,0 +1,162 @@ +"""Durable, streaming BugAnalysis intake client. Python stdlib only.""" +from contextlib import contextmanager +import hashlib +import http.client +import json +import os +from pathlib import Path +import re +import secrets +import sqlite3 +import time +from urllib.parse import urlsplit + + +class HandoffError(Exception): + pass + + +def settings(raw,base): + if raw and type(raw.get('enabled',False)) is not bool:raise HandoffError('HANDOFF_CONFIG') + if not raw or not raw.get('enabled',False):return None + value=dict(raw) + allowed={'enabled','url','project_id','profile_id','token_env','outbox','timeout_seconds','retry_seconds','producer_id'} + if set(value)-allowed:raise HandoffError('HANDOFF_CONFIG') + value.setdefault('profile_id','vcore');value.setdefault('token_env','VBA_INTAKE_TOKEN') + value.setdefault('timeout_seconds',120);value.setdefault('retry_seconds',60) + for field in ('project_id','profile_id','producer_id'): + if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,63}',str(value.get(field,''))):raise HandoffError('HANDOFF_CONFIG') + if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*',value['token_env']):raise HandoffError('HANDOFF_CONFIG') + parsed=urlsplit(value.get('url','')) + if (parsed.scheme not in {'https','http'} or not parsed.hostname or parsed.username or parsed.password + or parsed.query or parsed.fragment or parsed.path not in {'','/'} + or (parsed.scheme=='http' and parsed.hostname not in {'127.0.0.1','localhost','::1'})): + raise HandoffError('HANDOFF_URL') + if any(type(value[f]) is not int or not 1<=value[f]<=3600 for f in ('timeout_seconds','retry_seconds')): + raise HandoffError('HANDOFF_CONFIG') + value['outbox']=str((base/Path(value.get('outbox','buganalysis-outbox.sqlite3'))).resolve()) + return value + + +def digest(path): + h=hashlib.sha256() + with Path(path).open('rb') as stream: + for chunk in iter(lambda:stream.read(1024*1024),b''):h.update(chunk) + return h.hexdigest() + + +class Outbox: + def __init__(self,config,*,clock=time.time): + self.config=config;self.clock=clock;self.path=Path(config['outbox']) + self.path.parent.mkdir(parents=True,exist_ok=True) + if not self.path.exists(): + try: + fd=os.open(self.path,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600);os.close(fd) + except FileExistsError:pass + with self.db() as db: + db.execute('CREATE TABLE IF NOT EXISTS deliveries (job TEXT PRIMARY KEY, idempotency TEXT NOT NULL UNIQUE, payload TEXT NOT NULL,status TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, next_try REAL NOT NULL DEFAULT 0, code TEXT, receipt TEXT, lease_until REAL NOT NULL DEFAULT 0)') + + @contextmanager + def db(self): + conn=sqlite3.connect(self.path,timeout=10);conn.row_factory=sqlite3.Row + try: + with conn:yield conn + finally:conn.close() + + def stage(self,run_id,bug_id,diag,src,revision): + files=[] + for role,path in (('wowdiag',diag),('wowsrc',src)): + path=Path(path).resolve() + files.append({'role':role,'path':str(path),'size':path.stat().st_size,'sha256':digest(path)}) + # The adapters import actual observations/title from the unchanged archives. + # Do not invent an observation merely to populate envelope metadata. + metadata={'kind':'new_case','project_id':self.config['project_id'],'profile_id':self.config['profile_id'], + 'case_id':f'BUG-{bug_id:06d}','report':{'title':f'BUG-{bug_id:06d}'},'versions':{'capture_server':revision}} + payload=json.dumps({'url':self.config['url'],'metadata':metadata,'files':files},sort_keys=True) + job=f"{self.config['producer_id']}:{run_id}" + with self.db() as db: + db.execute('BEGIN IMMEDIATE') + row=db.execute('SELECT payload FROM deliveries WHERE job=?',(job,)).fetchone() + if row: + if row['payload']!=payload:raise HandoffError('HANDOFF_JOB_CONTENT_CHANGED') + return + db.execute('INSERT INTO deliveries(job,idempotency,payload,status) VALUES (?,?,?,?)', + (job,'capture-'+secrets.token_hex(24),payload,'pending')) + + def status(self): + with self.db() as db: + return [dict(r) for r in db.execute('SELECT job,status,attempts,next_try,code,receipt FROM deliveries ORDER BY job')] + + def flush(self,*,send=None,force=False,maximum=5): + send=send or upload + for _ in range(maximum): + now=self.clock() + with self.db() as db: + db.execute('BEGIN IMMEDIATE') + row=db.execute("SELECT * FROM deliveries WHERE status='pending' AND lease_until<=? AND (next_try<=? OR ?) ORDER BY next_try,job LIMIT 1",(now,now,int(force))).fetchone() + if not row:return + # Lease prevents concurrent --retry-handoffs and worker submissions. + db.execute('UPDATE deliveries SET lease_until=?,attempts=attempts+1 WHERE job=?', + (now+self.config['timeout_seconds']+60,row['job'])) + status='pending';receipt=None + try: + payload=json.loads(row['payload']) + for file in payload['files']: + path=Path(file['path']) + if not path.is_file() or path.stat().st_size!=file['size'] or digest(path)!=file['sha256']: + raise HandoffError('HANDOFF_ARCHIVE_CHANGED') + code,receipt=send(self.config,payload,row['idempotency']) + if code=='OK':status='sent' + elif code.startswith('HTTP_') and code not in {'HTTP_401','HTTP_403','HTTP_408','HTTP_425','HTTP_429','HTTP_500','HTTP_502','HTTP_503','HTTP_504','HTTP_507'}:status='attention' + except HandoffError as error: + code=str(error) + if code=='HANDOFF_ARCHIVE_CHANGED':status='attention' + except Exception: + # Never persist arbitrary exception text or HTTP bodies (may contain secrets). + code='HANDOFF_TRANSPORT' + with self.db() as db: + db.execute('UPDATE deliveries SET status=?,next_try=?,code=?,receipt=?,lease_until=0 WHERE job=?', + (status,self.clock()+self.config['retry_seconds'],code,json.dumps(receipt) if receipt else None,row['job'])) + print(json.dumps({'handoff':row['job'],'status':status,'code':code}),flush=True) + if status=='pending':return # avoid retrying this request repeatedly in the same pass + + +def upload(config,payload,idempotency): + token=os.environ.get(config['token_env'],'') + if not re.fullmatch(r'[A-Za-z0-9_-]{32,256}',token):raise HandoffError('HANDOFF_CREDENTIALS') + target=urlsplit(payload['url']);boundary='vba-'+secrets.token_hex(24) + prefix=(f'--{boundary}\r\nContent-Disposition: form-data; name="metadata"\r\n\r\n'+json.dumps(payload['metadata'])+'\r\n').encode() + parts=[] + for file in payload['files']: + role=file['role'] + header=f'--{boundary}\r\nContent-Disposition: form-data; name="{role}"; filename="{role}.tar.gz"\r\nContent-Type: application/gzip\r\n\r\n'.encode() + parts.append((header,file)) + ending=f'--{boundary}--\r\n'.encode() + length=len(prefix)+len(ending)+sum(len(header)+file['size']+2 for header,file in parts) + connection=(http.client.HTTPSConnection if target.scheme=='https' else http.client.HTTPConnection)(target.hostname,target.port,timeout=config['timeout_seconds']) + started=time.monotonic() + def transmit(chunk): + remaining=config['timeout_seconds']-(time.monotonic()-started) + if remaining<=0:raise HandoffError('HANDOFF_TIMEOUT') + if connection.sock:connection.sock.settimeout(remaining) + connection.send(chunk) + try: + connection.request('POST','/v1/submissions',headers={'Authorization':'Bearer '+token, + 'Idempotency-Key':idempotency,'Content-Type':'multipart/form-data; boundary='+boundary, + 'Content-Length':str(length)},body=None) + transmit(prefix) + for header,file in parts: + transmit(header) + with Path(file['path']).open('rb') as stream: + for chunk in iter(lambda:stream.read(1024*1024),b''):transmit(chunk) + transmit(b'\r\n') + transmit(ending) + response=connection.getresponse();body=response.read(65537) + if response.status not in {200,201}:return f'HTTP_{response.status}',None + if len(body)>65536:raise HandoffError('HANDOFF_RECEIPT_INVALID') + try:data=json.loads(body) + except ValueError:raise HandoffError('HANDOFF_RECEIPT_INVALID') from None + if (not isinstance(data,dict) or not isinstance(data.get('submission_id'),str) + or data.get('case_id')!=payload['metadata']['case_id']):raise HandoffError('HANDOFF_RECEIPT_INVALID') + return 'OK',{field:data[field] for field in ('submission_id','case_id','status') if field in data} + finally:connection.close() diff --git a/deploy/capture-worker/config.toml.example b/deploy/capture-worker/config.toml.example new file mode 100644 index 0000000..59755ff --- /dev/null +++ b/deploy/capture-worker/config.toml.example @@ -0,0 +1,29 @@ +[database] +host = "127.0.0.1" +port = 3306 +user = "vcore_worker" +password = "CHANGE_ME" +database = "vcore_dev" + +[worker] +poll_seconds = 10 +job_timeout_seconds = 900 + +[paths] +python = "/usr/bin/python3" +wowdiag_script = "/home/wow_admin/vcore-tools/wowdiag/wowdiag.py" +wowdiag_config = "/home/wow_admin/vcore-tools/wowdiag/config.toml" +wowsrc_script = "/home/wow_admin/vcore-tools/wowsrc/wowsrc.py" +wowsrc_config = "/home/wow_admin/vcore-tools/wowsrc/config.toml" +log_dir = "/home/wow_admin/vcore-data/diagnostics/worker-logs" + +[buganalysis] +enabled = false +url = "https://buganalysis.example.org" +project_id = "viano-core" +profile_id = "vcore" +producer_id = "vcore-testserver" +token_env = "VBA_INTAKE_TOKEN" +outbox = "buganalysis-outbox.sqlite3" +timeout_seconds = 120 +retry_seconds = 60 diff --git a/deploy/capture-worker/wowdiag_worker.py b/deploy/capture-worker/wowdiag_worker.py new file mode 100644 index 0000000..844fb1b --- /dev/null +++ b/deploy/capture-worker/wowdiag_worker.py @@ -0,0 +1,617 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import re +import socket +import subprocess +import sys +import time +import tomllib +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +import pymysql +from buganalysis_handoff import Outbox, settings as handoff_settings + +WORKER_VERSION = "0.4" +ARCHIVE_RE = re.compile(r"^Archive created:\s*(.+?)\s*$", re.MULTILINE) +WOWDIAG_PACKAGE_RE = re.compile(r"^Diagnostic package created:\s*(.+?)\s*$", re.MULTILINE) +WOWSRC_PACKAGE_RE = re.compile(r"^wowsrc package created:\s*(.+?)\s*$", re.MULTILINE) + + +@dataclass(frozen=True) +class Config: + host: str + port: int + user: str + password: str + database: str + poll_seconds: int + timeout: int + python: str + wowdiag_script: Path + wowdiag_config: Path + wowsrc_script: Path + wowsrc_config: Path + log_dir: Path + buganalysis: Optional[dict] = None + + +@dataclass(frozen=True) +class ToolContext: + diagnostics_root: Path + source_context_root: Path + core_source: Path + + +@dataclass(frozen=True) +class Job: + run_id: int + bug_id: int + claim_token: str + + +def load_config(path: Path) -> Config: + with path.open("rb") as f: + d = tomllib.load(f) + + db, w, p = d["database"], d.get("worker", {}), d["paths"] + return Config( + str(db.get("host", "127.0.0.1")), + int(db.get("port", 3306)), + str(db["user"]), + str(db["password"]), + str(db.get("database", "vcore_dev")), + max(1, int(w.get("poll_seconds", 10))), + max(30, int(w.get("job_timeout_seconds", 900))), + str(p.get("python", sys.executable)), + Path(p["wowdiag_script"]).resolve(), + Path(p["wowdiag_config"]).resolve(), + Path(p["wowsrc_script"]).resolve(), + Path(p["wowsrc_config"]).resolve(), + Path(p["log_dir"]).resolve(), + handoff_settings(d.get('buganalysis'),path.parent), + ) + + +def load_tool_context(cfg: Config) -> ToolContext: + with cfg.wowdiag_config.open("rb") as f: + diag = tomllib.load(f) + with cfg.wowsrc_config.open("rb") as f: + src = tomllib.load(f) + + diag_paths = diag["paths"] + src_paths = src["paths"] + + diag_core = Path(diag_paths["core_source"]).resolve() + src_core = Path(src_paths["core_source"]).resolve() + if diag_core != src_core: + raise RuntimeError( + "wowdiag and wowsrc use different core_source paths: " + f"wowdiag={diag_core}, wowsrc={src_core}" + ) + + return ToolContext( + diagnostics_root=Path(diag_paths["diagnostics"]).resolve(), + source_context_root=Path(src_paths["output_root"]).resolve(), + core_source=diag_core, + ) + + +def connect(cfg: Config): + return pymysql.connect( + host=cfg.host, + port=cfg.port, + user=cfg.user, + password=cfg.password, + database=cfg.database, + charset="utf8mb4", + autocommit=False, + cursorclass=pymysql.cursors.DictCursor, + ) + + +def identity() -> str: + return f"{socket.gethostname()}:{os.getpid()}" + + +def claim_one(conn, claimed_by: str) -> Optional[Job]: + token = uuid.uuid4().hex + try: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE diagnostic_runs + SET status='running', claim_token=%s, claimed_by=%s, + claimed_at=NOW(), started_at=NOW(), error_message=NULL + WHERE status='queued' + ORDER BY diagnostic_run_id ASC + LIMIT 1 + """, + (token, claimed_by), + ) + if cur.rowcount != 1: + conn.rollback() + return None + + conn.commit() + + with conn.cursor() as cur: + cur.execute( + """SELECT diagnostic_run_id, bug_id, claim_token + FROM diagnostic_runs + WHERE claim_token=%s AND status='running' LIMIT 1""", + (token,), + ) + row = cur.fetchone() + + if not row: + raise RuntimeError("Claim succeeded but claimed row could not be resolved") + + return Job( + int(row["diagnostic_run_id"]), + int(row["bug_id"]), + str(row["claim_token"]), + ) + except Exception: + conn.rollback() + raise + + +def mark_completed( + conn, + job: Job, + wowdiag_archive: Path, + wowdiag_version: str, + core_revision: str, +): + """ + Keep the existing diagnostic_runs schema contract intact. + + output_path remains the wowdiag archive. The paired wowsrc archive is + generated in the configured wowsrc output_root and is recorded in the + worker log. A future DB migration can add dedicated wowsrc columns without + changing the pipeline execution semantics introduced here. + """ + with conn.cursor() as cur: + cur.execute( + """ + UPDATE diagnostic_runs + SET status='completed', finished_at=NOW(), output_path=%s, + error_message=NULL, wowdiag_version=%s, core_revision=%s + WHERE diagnostic_run_id=%s AND claim_token=%s AND status='running' + """, + ( + str(wowdiag_archive), + wowdiag_version[:64], + core_revision[:64], + job.run_id, + job.claim_token, + ), + ) + if cur.rowcount != 1: + conn.rollback() + raise RuntimeError("Could not finalize job: claim ownership/status changed") + conn.commit() + + +def mark_failed(conn, job: Job, message: str): + with conn.cursor() as cur: + cur.execute( + """ + UPDATE diagnostic_runs + SET status='failed', finished_at=NOW(), error_message=%s + WHERE diagnostic_run_id=%s AND claim_token=%s AND status='running' + """, + (message[:65535], job.run_id, job.claim_token), + ) + if cur.rowcount != 1: + conn.rollback() + raise RuntimeError("Could not mark job failed: claim ownership/status changed") + conn.commit() + + +def validated_path_from_match( + stdout: str, + regex: re.Pattern[str], + root: Path, + description: str, + must_be_dir: bool, +) -> Optional[Path]: + m = regex.search(stdout) + if not m: + return None + + p = Path(m.group(1).strip()).resolve() + try: + p.relative_to(root) + except ValueError as exc: + raise RuntimeError(f"{description} outside configured output root: {p}") from exc + + if must_be_dir: + if not p.is_dir(): + raise RuntimeError(f"{description} directory does not exist: {p}") + elif not p.is_file(): + raise RuntimeError(f"{description} file does not exist: {p}") + + return p + + +def package_from_stdout( + stdout: str, + regex: re.Pattern[str], + root: Path, + description: str, +) -> Optional[Path]: + return validated_path_from_match( + stdout, regex, root, description, must_be_dir=True + ) + + +def archive_from_stdout(stdout: str, root: Path, description: str) -> Optional[Path]: + p = validated_path_from_match( + stdout, ARCHIVE_RE, root, description, must_be_dir=False + ) + if p and p.suffixes[-2:] != [".tar", ".gz"]: + raise RuntimeError(f"unexpected archive type: {p}") + return p + + +def fallback_archive(root: Path, pattern: str, started: float) -> Optional[Path]: + found = [] + for p in root.glob(pattern): + try: + st = p.stat() + except OSError: + continue + if st.st_mtime >= started - 2: + found.append((st.st_mtime, p.resolve())) + return sorted(found, reverse=True)[0][1] if found else None + + +def manifest_metadata(package_dir: Path, expected_tool: str) -> tuple[str, str]: + manifest_path = package_dir / "manifest.json" + if not manifest_path.is_file(): + raise RuntimeError(f"{expected_tool} manifest not found: {manifest_path}") + + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except Exception as exc: + raise RuntimeError(f"Could not read {expected_tool} manifest: {exc}") from exc + + tool = str(manifest.get("tool") or "").strip() + if tool and tool != expected_tool: + raise RuntimeError( + f"Unexpected manifest tool in {manifest_path}: {tool!r}" + ) + + version = str(manifest.get("tool_version") or "").strip() + revision = str(manifest.get("core_revision") or "").strip() + + if not version: + raise RuntimeError(f"{expected_tool} manifest contains no tool_version") + if not revision: + raise RuntimeError(f"{expected_tool} manifest contains no core_revision") + + return version, revision + + +def run_command(cmd: list[str], cwd: Path, timeout: int) -> tuple[int, str, str]: + r = subprocess.run( + cmd, + cwd=str(cwd), + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + return r.returncode, r.stdout or "", r.stderr or "" + + +def write_log( + cfg: Config, + job: Job, + diag_cmd: list[str], + diag_rc: Optional[int], + diag_out: str, + diag_err: str, + src_cmd: Optional[list[str]] = None, + src_rc: Optional[int] = None, + src_out: str = "", + src_err: str = "", + wowdiag_archive: Optional[Path] = None, + wowsrc_archive: Optional[Path] = None, + failure: Optional[BaseException] = None, +): + cfg.log_dir.mkdir(parents=True, exist_ok=True) + p = cfg.log_dir / f"DIAG-{job.run_id:06d}.log" + + parts = [ + f"worker_version={WORKER_VERSION}", + f"diagnostic_run_id={job.run_id}", + f"bug_id={job.bug_id}", + f"claim_token={job.claim_token}", + f"wowdiag_archive={wowdiag_archive or ''}", + f"wowsrc_archive={wowsrc_archive or ''}", + "", + "WOWDIAG COMMAND:", + repr(diag_cmd), + f"returncode={diag_rc}", + "", + "WOWDIAG STDOUT:", + diag_out or "", + "", + "WOWDIAG STDERR:", + diag_err or "", + ] + + if src_cmd is not None: + parts += [ + "", + "WOWSRC COMMAND:", + repr(src_cmd), + f"returncode={src_rc}", + "", + "WOWSRC STDOUT:", + src_out or "", + "", + "WOWSRC STDERR:", + src_err or "", + ] + + if failure: + parts += ["", "ERROR:", str(failure)] + + p.write_text("\n".join(parts), encoding="utf-8") + + +def execute_pipeline(cfg: Config, ctx: ToolContext, job: Job): + if not cfg.wowdiag_script.is_file(): + raise RuntimeError(f"wowdiag script not found: {cfg.wowdiag_script}") + if not cfg.wowdiag_config.is_file(): + raise RuntimeError(f"wowdiag config not found: {cfg.wowdiag_config}") + if not cfg.wowsrc_script.is_file(): + raise RuntimeError(f"wowsrc script not found: {cfg.wowsrc_script}") + if not cfg.wowsrc_config.is_file(): + raise RuntimeError(f"wowsrc config not found: {cfg.wowsrc_config}") + + ctx.diagnostics_root.mkdir(parents=True, exist_ok=True) + ctx.source_context_root.mkdir(parents=True, exist_ok=True) + + diag_cmd = [ + cfg.python, + str(cfg.wowdiag_script), + "--config", + str(cfg.wowdiag_config), + "bug", + str(job.bug_id), + ] + + src_cmd: Optional[list[str]] = None + diag_out = diag_err = src_out = src_err = "" + diag_rc: Optional[int] = None + src_rc: Optional[int] = None + wowdiag_archive: Optional[Path] = None + wowsrc_archive: Optional[Path] = None + + try: + diag_started = time.time() + diag_rc, diag_out, diag_err = run_command( + diag_cmd, cfg.wowdiag_script.parent, cfg.timeout + ) + if diag_rc != 0: + raise RuntimeError( + f"wowdiag exited with code {diag_rc}; " + f"stderr: {diag_err.strip() or ''}" + ) + + diag_package = package_from_stdout( + diag_out, + WOWDIAG_PACKAGE_RE, + ctx.diagnostics_root, + "wowdiag package", + ) + if not diag_package: + raise RuntimeError( + "wowdiag exited successfully but no package directory could be identified" + ) + + wowdiag_archive = ( + archive_from_stdout(diag_out, ctx.diagnostics_root, "wowdiag archive") + or fallback_archive( + ctx.diagnostics_root, + f"BUG-{job.bug_id:06d}-*.tar.gz", + diag_started, + ) + ) + if not wowdiag_archive: + raise RuntimeError( + "wowdiag exited successfully but no archive could be identified" + ) + + wowdiag_version, diag_revision = manifest_metadata(diag_package, "wowdiag") + + # wowsrc deliberately consumes the exact wowdiag archive generated by + # this same queue job. This prevents it from silently picking up an + # older package for the same bug. + src_cmd = [ + cfg.python, + str(cfg.wowsrc_script), + "--config", + str(cfg.wowsrc_config), + "bug", + str(job.bug_id), + "--diag", + str(wowdiag_archive), + ] + + src_started = time.time() + src_rc, src_out, src_err = run_command( + src_cmd, cfg.wowsrc_script.parent, cfg.timeout + ) + if src_rc != 0: + raise RuntimeError( + f"wowsrc exited with code {src_rc}; " + f"stderr: {src_err.strip() or ''}" + ) + + src_package = package_from_stdout( + src_out, + WOWSRC_PACKAGE_RE, + ctx.source_context_root, + "wowsrc package", + ) + if not src_package: + raise RuntimeError( + "wowsrc exited successfully but no package directory could be identified" + ) + + wowsrc_archive = ( + archive_from_stdout(src_out, ctx.source_context_root, "wowsrc archive") + or fallback_archive( + ctx.source_context_root, + f"WOWSRC-BUG-{job.bug_id:06d}-*.tar.gz", + src_started, + ) + ) + if not wowsrc_archive: + raise RuntimeError( + "wowsrc exited successfully but no archive could be identified" + ) + + _, src_revision = manifest_metadata(src_package, "wowsrc") + if diag_revision != src_revision: + raise RuntimeError( + "wowdiag and wowsrc were generated from different core revisions: " + f"wowdiag={diag_revision}, wowsrc={src_revision}" + ) + + write_log( + cfg, + job, + diag_cmd, + diag_rc, + diag_out, + diag_err, + src_cmd, + src_rc, + src_out, + src_err, + wowdiag_archive, + wowsrc_archive, + ) + + return wowdiag_archive, wowsrc_archive, wowdiag_version, diag_revision + + except Exception as exc: + write_log( + cfg, + job, + diag_cmd, + diag_rc, + diag_out, + diag_err, + src_cmd, + src_rc, + src_out, + src_err, + wowdiag_archive, + wowsrc_archive, + exc, + ) + raise + + +def process_one(cfg: Config, ctx: ToolContext) -> bool: + conn = connect(cfg) + try: + job = claim_one(conn, identity()) + if not job: + return False + + print( + f"Claimed DIAG-{job.run_id} for BUG-{job.bug_id} as {identity()}", + flush=True, + ) + + try: + diag_archive, src_archive, version, revision = execute_pipeline( + cfg, ctx, job + ) + if cfg.buganalysis: + Outbox(cfg.buganalysis).stage(job.run_id,job.bug_id,diag_archive,src_archive,revision) + mark_completed(conn, job, diag_archive, version, revision) + print( + f"Completed DIAG-{job.run_id}: " + f"wowdiag={diag_archive} wowsrc={src_archive}", + flush=True, + ) + except Exception as exc: + mark_failed(conn, job, str(exc)) + print( + f"FAILED DIAG-{job.run_id}: {exc}", + file=sys.stderr, + flush=True, + ) + + return True + finally: + conn.close() + + +def main() -> int: + ap = argparse.ArgumentParser( + prog="wowdiag-worker", + description="Vcore diagnostic_runs -> wowdiag -> wowsrc worker", + ) + ap.add_argument( + "--config", + default=str(Path(__file__).with_name("config.toml")), + ) + ap.add_argument( + "--once", + action="store_true", + help="Process at most one queued job and exit", + ) + ap.add_argument('--retry-handoffs',action='store_true',help='Retry queued API deliveries only; no capture jobs') + ap.add_argument('--handoff-status',action='store_true',help='List API delivery status; no capture jobs') + args = ap.parse_args() + + cfg = load_config(Path(args.config).resolve()) + outbox=Outbox(cfg.buganalysis) if cfg.buganalysis else None + if args.handoff_status or args.retry_handoffs: + if not outbox: + print('BugAnalysis handoff is disabled.',file=sys.stderr);return 1 + if args.retry_handoffs:outbox.flush(force=True,maximum=100) + print(json.dumps(outbox.status()));return 0 + ctx = load_tool_context(cfg) + + print(f"Vcore diagnostic worker v{WORKER_VERSION} ({identity()})", flush=True) + + if args.once: + if outbox:outbox.flush() + if not process_one(cfg, ctx): + print("No queued diagnostic job found.", flush=True) + if outbox:outbox.flush() + return 0 + + while True: + try: + if outbox:outbox.flush() + if not process_one(cfg, ctx): + time.sleep(cfg.poll_seconds) + except KeyboardInterrupt: + print("Worker stopped.", flush=True) + return 0 + except Exception as exc: + print(f"Worker loop error: {exc}", file=sys.stderr, flush=True) + time.sleep(cfg.poll_seconds) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/test-results/capture-worker-handoff-20261023.md b/docs/test-results/capture-worker-handoff-20261023.md new file mode 100644 index 0000000..2212f2b --- /dev/null +++ b/docs/test-results/capture-worker-handoff-20261023.md @@ -0,0 +1,37 @@ +# wowdiag-worker-Anschluss – 2026-10-23 + +Ausgangslage: Nutzer stellte `Z:/Dokumente/Projekte/wowdiag-worker/` bereit. +Quellcode bestätigt WORKER_VERSION 0.3; README/Beispielkonfiguration waren älter. +Die echte Konfiguration wurde nicht gelesen oder geändert. Keine echten Capture- +Jobs, Testserverabfragen, Cloudanfragen oder Dienststarts. + +Ausbau v0.4 unter `deploy/capture-worker`: unveränderte Claim-Verarbeitung, +optionale API-Übergabe nach erfolgreicher Erfassung, lokale SQLite-Outbox, +Streaming-Upload beider Archive, feste Idempotenz über Wiederholungen/Neustarts, +Inhaltsprüfung, begrenzte Requests, HTTPS für entfernte Hosts, keine Redirects, +keine Secrets in Quittungen/Logs, separate Status-/Retry-Befehle. +Die tatsächlichen Testerbeobachtungen bleiben in den Originalarchiven. + +Lokale Prüfungen: + +```text +python -m pytest tests/profiles/vcore/test_capture_handoff.py tests/profiles/vcore/test_intake_api.py tests/test_integration_keys.py -q +``` + +Prüffälle: URL-Konfiguration, persistente Outbox, Wiederholungsidentität, +Parallel-Lease, Archivänderung, Wiederholung nach verlorener Erfolgsantwort +gegen echte lokale Uvicorn-/Intake-API mit synthetischen Archiven sowie +Staging vor Capture-Abschluss und deaktivierter Übergabe. +15 Tests bestanden; CLI-Hilfe und `git diff --check` ebenfalls erfolgreich. +Feldprüfung mit tatsächlichem Erfassungsserver +steht aus. Keine Token-/Cache-/Abrechnungsmetriken, da keine LLM-Aufrufe. + +Speicher vor Tests: Trials 16.939.129.428 Bytes, Fehlerkorpus 2.471.041 Bytes, +pytest-Temporärdateien ca. 4,2 MB. Nur kleine synthetische Archive erzeugt; +keine vollständigen Bug-Paketarchive kopiert. Unterhalb des 20-GB-Budgets. + +Nach erfolgreichen Prüfungen wurde v0.4 mit Modul, Anleitung und Beispieldateien +in den vom Nutzer genannten Ordner auf Z: zurückgeschrieben. Vorhandene +Quelldatei, README und Beispielkonfiguration wurden jeweils als +`.before_api_v0_4` gesichert. Die aktive `config.toml` blieb unverändert. +Kein externer Dienst gestartet; Übergabe ist erst nach bewusster Konfiguration aktiv. diff --git a/tests/profiles/vcore/test_capture_handoff.py b/tests/profiles/vcore/test_capture_handoff.py new file mode 100644 index 0000000..c53324b --- /dev/null +++ b/tests/profiles/vcore/test_capture_handoff.py @@ -0,0 +1,103 @@ +import importlib.util +import json +from pathlib import Path +import socket +import sys +import threading +import time +from types import SimpleNamespace + +import pytest +import uvicorn + +from viano_bug_analysis.api import APIConfig,Client,Project,IntakeLimits +from viano_bug_analysis.api.app import create_app + +ROOT=Path(__file__).resolve().parents[3] +spec=importlib.util.spec_from_file_location('buganalysis_handoff',ROOT/'deploy/capture-worker/buganalysis_handoff.py') +bridge=importlib.util.module_from_spec(spec);spec.loader.exec_module(bridge) +sys.modules['buganalysis_handoff']=bridge + + +def config(tmp_path,url='http://127.0.0.1:8080'): + return bridge.settings({'enabled':True,'url':url,'project_id':'core','producer_id':'test-capture','outbox':'outbox.sqlite3'},tmp_path) + + +def test_settings_reject_credential_leak_paths(tmp_path): + assert bridge.settings({},tmp_path) is None + for url in ('http://external.example','https://user:password@example.org','https://example.org/path','https://example.org?secret=x'): + with pytest.raises(bridge.HandoffError):config(tmp_path,url) + + +def test_outbox_retries_integrity_and_parallel_lease(tmp_path,source_pair): + now=[10.0];cfg=config(tmp_path);outbox=bridge.Outbox(cfg,clock=lambda:now[0]) + diag,src,*_=source_pair + outbox.stage(1,20,diag,src,'revision');outbox.stage(1,20,diag,src,'revision') + assert len(outbox.status())==1 + seen=[] + def failed(cfg,payload,key):seen.append(key);raise OSError('secret must not persist') + outbox.flush(send=failed) + assert outbox.status()[0]['status']=='pending' + assert 'secret' not in json.dumps(outbox.status()) + other=bridge.Outbox(cfg,clock=lambda:now[0]) + def successful(cfg,payload,key): + seen.append(key) + other.flush(force=True,send=lambda *_:pytest.fail('Concurrent lease ignored')) + return 'OK',{'submission_id':'id'} + now[0]+=61;outbox.flush(send=successful) + assert len(set(seen))==1 and outbox.status()[0]['status']=='sent' + outbox.stage(2,20,diag,src,'revision') + diag.write_bytes(b'changed') + outbox.flush(send=lambda *_:pytest.fail('Changed archives uploaded')) + assert outbox.status()[1]['code']=='HANDOFF_ARCHIVE_CHANGED' + + +def test_real_streaming_intake_and_lost_response_replay(tmp_path,source_pair,monkeypatch): + token='test_token_'+'x'*40;monkeypatch.setenv('VBA_INTAKE_TOKEN',token) + api=APIConfig(tmp_path/'service',(Project('core','wow','vcore'),), + (Client('capture','VBA_INTAKE_TOKEN',('core',)),),limits=IntakeLimits(min_free_bytes=1)) + app=create_app(api) + sock=socket.socket();sock.bind(('127.0.0.1',0));sock.listen() + port=sock.getsockname()[1] + server=uvicorn.Server(uvicorn.Config(app,log_level='critical',access_log=False)) + thread=threading.Thread(target=lambda:server.run(sockets=[sock]),daemon=True);thread.start() + try: + for _ in range(100): + if server.started:break + time.sleep(.02) + assert server.started + cfg=config(tmp_path,f'http://127.0.0.1:{port}') + outbox=bridge.Outbox(cfg) + outbox.stage(1,20,source_pair[0],source_pair[1],'verified-source-revision') + def lose_response(*args): + assert bridge.upload(*args)[0]=='OK' + raise OSError('Response lost after acceptance') + outbox.flush(send=lose_response) + assert outbox.status()[0]['status']=='pending' + bridge.Outbox(cfg).flush(force=True) + row=outbox.status()[0] + assert row['status']=='sent' and json.loads(row['receipt'])['case_id']=='BUG-000020' + assert len(list((tmp_path/'service/runtime/intake/requests').rglob('*.json')))==1 + assert token.encode() not in outbox.path.read_bytes() + finally: + server.should_exit=True;thread.join(10);sock.close() + assert not thread.is_alive() + + +def test_worker_stages_before_capture_completion_and_can_disable(tmp_path,source_pair,monkeypatch): + worker_spec=importlib.util.spec_from_file_location('capture_worker_v04',ROOT/'deploy/capture-worker/wowdiag_worker.py') + worker=importlib.util.module_from_spec(worker_spec);sys.modules[worker_spec.name]=worker;worker_spec.loader.exec_module(worker) + closed=[];completed=[] + monkeypatch.setattr(worker,'connect',lambda cfg:SimpleNamespace(close=lambda:closed.append(True))) + monkeypatch.setattr(worker,'claim_one',lambda *_:worker.Job(1,20,'claim')) + monkeypatch.setattr(worker,'execute_pipeline',lambda *_:(source_pair[0],source_pair[1],'version','revision')) + cfg=SimpleNamespace(buganalysis=config(tmp_path)) + def complete(*args): + if cfg.buganalysis:assert bridge.Outbox(cfg.buganalysis).status()[0]['status']=='pending' + completed.append(True) + monkeypatch.setattr(worker,'mark_completed',complete) + monkeypatch.setattr(worker,'mark_failed',lambda *_:pytest.fail('Successful capture marked failed')) + assert worker.process_one(cfg,None) + cfg.buganalysis=None + assert worker.process_one(cfg,None) + assert len(completed)==len(closed)==2