diff --git a/README.md b/README.md index 84e4e78..bc9a832 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,8 @@ Vcore imports for wowdiag/wowsrc are documented in 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. +The [continuous analysis worker](docs/analysis-worker.md) consumes that inbox, +routes cases to configured projects and resumes checkpointed analysis after restart. Requires Python 3.11 or newer. @@ -29,8 +31,7 @@ per-hypothesis budget accounting. An LLM-driven investigation controller connect read-only queries, durable evidence and asynchronous test handoffs. Final evaluation produces traceable diagnoses or human review. Confirmed diagnoses can produce policy-bound Change Requests, with final output validation and atomic local -outbox/review publication. Opt-in operational telemetry observes the existing APIs. A continuous worker -remains a future milestone. +outbox/review publication. Operational telemetry observes the APIs and continuous worker. A bounded single-case command-line trial now connects import, analysis and local delivery with telemetry: [first real LLM trial](docs/first-llm-trial.md). diff --git a/config/api.example.toml b/config/api.example.toml index 62c433c..0d9c4d5 100644 --- a/config/api.example.toml +++ b/config/api.example.toml @@ -8,6 +8,7 @@ id = "viano-core" type = "wow" profile = "vcore" adapter = "vcore" +planning_mode = "joint" # Trusted server-side analysis settings; never accepted from upload metadata. # [projects.settings] # project_config = "project.local.toml" diff --git a/config/worker.example.toml b/config/worker.example.toml new file mode 100644 index 0000000..3f0e66a --- /dev/null +++ b/config/worker.example.toml @@ -0,0 +1,8 @@ +[worker] +# Same intake configuration/workspace/project catalogue as the HTTP API. +intake_config = "api.local.toml" +poll_seconds = 5 +error_retry_seconds = 60 +max_inbox_per_pass = 20 +min_free_bytes = 536870912 +# telemetry_config = "telemetry.local.toml" diff --git a/deploy/viano-bug-analysis-worker.service b/deploy/viano-bug-analysis-worker.service new file mode 100644 index 0000000..f94cef9 --- /dev/null +++ b/deploy/viano-bug-analysis-worker.service @@ -0,0 +1,22 @@ +[Unit] +Description=Viano BugAnalysis analysis worker +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=buganalysis +Group=buganalysis +WorkingDirectory=/opt/viano-bug-analysis +EnvironmentFile=/etc/viano-bug-analysis/worker.env +ExecStart=/opt/viano-bug-analysis/.venv/bin/python -m viano_bug_analysis.worker run --config /opt/viano-bug-analysis/config/worker.local.toml +Restart=on-failure +RestartSec=10 +KillSignal=SIGTERM +# Finish the current checkpointed dispatch; operator can still force-stop. +TimeoutStopSec=infinity +UMask=0077 +NoNewPrivileges=true + +[Install] +WantedBy=multi-user.target diff --git a/docs/analysis-worker.md b/docs/analysis-worker.md new file mode 100644 index 0000000..3ab8330 --- /dev/null +++ b/docs/analysis-worker.md @@ -0,0 +1,169 @@ +# Dauerhafter Analyseworker + +Der Worker verbindet die Übergabe-API mit dem vorhandenen Analyseablauf. Er nimmt +Pakete aus derselben Inbox auf, wählt die serverseitige Projektkonfiguration und +bearbeitet aktive Fälle bis zu Outbox oder Review. Er erstellt keine Patches, +führt keine Builds aus und verändert keine Projekt-Datenbank. + +## Konfiguration + +`config/worker.example.toml` verweist auf die Konfiguration der Übergabe-API. +API und Worker müssen denselben Workspace und Projektkatalog verwenden. Der +Worker benötigt die API-Clienttokens nicht, wohl aber die in seiner LLM-Konfiguration +genannten Provider-Zugangsdaten. Beide Prozesse dürfen getrennt laufen. + +Für jedes freigegebene Projekt werden die Analysekonfigurationen in der +API-Konfiguration hinterlegt: + +```toml +[[projects]] +id = "viano-core" +type = "wow" +profile = "vcore" +adapter = "vcore" +planning_mode = "joint" +[projects.settings] +project_config = "project.local.toml" +llm_config = "llm.early-review.local.toml" +investigation_config = "investigation.bounded.toml" +``` + +Die Pfade sind relativ zur API-Konfigurationsdatei. `llm_config` ist erforderlich. +Weitere optionale Einstellungen: `analysis_config`, `budget_config`, +`diagnosis_config`, `change_request_config`, `correction_config`. Ohne explizite +Datei gelten deren bestehende Standardwerte. Für den aktuellen Vcore-Betrieb die +erprobte `investigation.bounded.toml` verwenden; ältere ausführliche Promptmodi +können an der unveränderten 20.000-Zeichen-Grenze scheitern. + +`planning_mode` ist je Projekt `joint` (Standard) oder `separate`. Die Profilbindung +einer Akte muss vollständig zum Katalog passen: `project.id`, `type`, `profile`. +Alte manuelle Pakete ohne `project.id` werden nicht automatisch zugeordnet. +Andere Projektadapter bleiben ein eigener Ausbauschritt. + +## Start und Prüfung + +```powershell +.\.venv\Scripts\python.exe -m viano_bug_analysis.worker check --config config/worker.local.toml +.\.venv\Scripts\python.exe -m viano_bug_analysis.worker run --config config/worker.local.toml +``` + +`check` lädt Konfiguration und prüft die Anwesenheit der LLM-Zugangsdaten. Es sendet +keine Modellanfrage und ist kein Erreichbarkeitstest für API, Datenbank oder +Repository. `run` verarbeitet angenommene Aufträge und kann deshalb kostenpflichtige +LLM-Aufrufe auslösen. Der reine Start ohne anstehende Fälle führt keine Modellaufrufe +aus. `run --once` führt genau einen Scheduler-Durchgang aus, nicht einen vollständigen +Buglauf. Auch dieser Modus kann LLM-Aufrufe auslösen. + +API zusätzlich in einem zweiten Prozess starten: + +```powershell +.\.venv\Scripts\python.exe -m viano_bug_analysis.api serve --config config/api.local.toml +``` + +Für diese Entwicklung wurden die zuvor nicht vorhandenen lokalen Dateien +`config/api.local.toml` und `config/worker.local.toml` mit dem Vcore-Profil vorbereitet. +Sie sind nicht versioniert und enthalten keine Zugangsschlüssel. Vorhandene lokale +Konfigurationen werden bei späteren Änderungen nicht automatisch überschrieben. + +## Scheduling und Wiederanlauf + +- Pro Durchgang erhält jeder berechtigte aktive Fall einen Dispatch. Danach + darf der nächste Fall arbeiten. Gibt es Fortschritt, folgt der nächste Durchgang + unmittelbar; nur im Leerlauf oder bei Wartefällen greift das Pollintervall. +- Der Dienst läuft als einzelner Analyseprozess. Eine OS-Dateisperre verhindert + einen zweiten Worker im selben Workspace. Die Fall-Lease ist mit dem manuellen + Trialbefehl geteilt. API-Veröffentlichungen verwenden weiterhin die kurzen + Checkpoint-/Inbox-Sperren; während einer Modellanfrage bleibt die API benutzbar. +- Paketmanifest und Checkpoints bleiben maßgeblich. Es wird nur die neueste + Generation pro Fall geprüft, nicht jede historische Generation erneut geöffnet. +- Bei Neustart werden verwaltete aktive Fälle und noch nicht übernommene Uploads + wiedergefunden. PREPARED-Veröffentlichungen der API werden vor der Verarbeitung + abgeschlossen. Rückläufer starten die bereits eröffnete neue Analyse-Revision; + alte Interpretation und Hypothesen werden nicht als neue Ergebnisse behandelt. +- Bestehende LLM-Operationskennungen bleiben stabil. Der Dienst ändert keine + Versuchskennungen, Budgets, Iterationsgrenzen oder Korrekturregeln, um einen + blockierten Fall weiterzutreiben. +- `RETRY_PENDING`/`DEFERRED` beachten die gespeicherte Wartezeit. Ein unerwarteter + Konfigurations-/Vorbereitungsfehler erhält einen lokalen Cooldown; andere Fälle + werden weiterbearbeitet. Eine ausdrücklich zurückgegebene `ERROR`-Antwort oder + unerwartete Exception während eines Dispatch erzeugt `ATTENTION`, statt dieselbe + Fehlantwort endlos abzuspielen oder möglicherweise erneut Modellkosten auszulösen. +- BLOCKED bleibt BLOCKED, wenn eine menschliche Testerantwort benötigt wird, + automatische Wiederholung untersagt ist, Modellvalidierung gescheitert ist oder + die LLM-Wiederholungen ausgeschöpft sind. Explizit freigegebene, fällige technische + Wiederholungen dürfen über den vorhandenen Resume-Checkpoint fortgesetzt werden. +- Outbox-/Review-Veröffentlichung wird nach Erfolg nicht ständig wiederholt. Nach + einem Absturz zwischen Paketabschluss und Veröffentlichung kann der bestehende + Outputvalidator die fehlende Veröffentlichung idempotent nachholen. + +Ein Prozessabbruch während einer externen Modellanfrage kann deren bereits +berechnete Kosten nicht zurücknehmen. Ist die Antwort vor dem Checkpoint verloren +gegangen, lässt sich eine erneute Anfrage nicht grundsätzlich ausschließen. Der +Worker verspricht keine Exactly-once-Ausführung beim externen Anbieter. + +## Betrieb und Fehlerbehandlung + +```powershell +.\.venv\Scripts\python.exe -m viano_bug_analysis.worker status --config config/worker.local.toml +``` + +`runtime/worker/status.json` enthält den letzten Dienstzustand und gegebenenfalls +den bearbeiteten Fall. Es ist ein Snapshot, kein Beweis für einen noch lebenden +Prozess. Die bestehende Telemetrie liefert ergänzend Heartbeat, Phasen, Laufzeiten +und Modellverbrauch. `runtime/worker/cases/` enthält kleine lokale +Schedulingquittungen mit Status, Fehlercode, Wartezeit und Konfigurationshashes. +Keine Prompts, Antworten oder Geheimnisse werden diesen Quittungen hinzugefügt. +Die Statusantwort der Übergabe-API enthält für einen zugeordneten Fall zusätzlich +den letzten Workerstatus. + +Die Akte bleibt vollständig transportabel. Die lokalen Quittungen sind keine +fachliche Beweisquelle und keine verteilte Jobdatenbank. Bei ihrer vollständigen +Entfernung müssen reine Scheduler-Cooldowns/Attention-Markierungen gegebenenfalls +neu erkannt werden; fachliche Sperren und LLM-Operationshistorie bleiben im Paket. + +Konfigurationsdateien werden über ihre Hashes vor dem ersten Dispatch gebunden. +Eine geänderte Konfiguration wird nicht unbemerkt auf dieselbe aktive Revision +angewendet. Neue Profileinstellungen benötigen einen Worker-Neustart; laufende +Revisionen mit abweichender Bindung gehen auf `ATTENTION/WORKER_CONFIG_CHANGED`. +Auch abhängige Profil-/Relationsdateien nicht während einer Untersuchung ändern. +Eine allgemeine Migration laufender Modelloperationen auf neue Policies ist noch +kein Bestandteil dieses Dienstes. + +Nach Behebung eines lokalen Schedulingfehlers kann der Betreiber den **gestoppten** +Worker freigeben: + +```powershell +.\.venv\Scripts\python.exe -m viano_bug_analysis.worker retry --config config/worker.local.toml --case-id BUG-000020 --package-id api-PAKETIDENTITAET +``` + +Dies entfernt die lokale Warte-/Attention-Markierung, aber setzt keine LLM-Operation +oder fachliche Budgets zurück. Für einen fälligen `RESEARCH_UNAVAILABLE`-Fall +bestätigt der Befehl zusätzlich die Reparatur des Research-Dienstes und setzt dessen +Resume-Checkpoint fort. Andere BLOCKED-Gründe, ausstehende menschliche Rückmeldungen +und abweichende Konfigurationsbindungen werden damit nicht umgangen. + +Bei zu wenig freiem Speicher pausiert der Worker mit `STORAGE_LOW`, ohne Akten zu +löschen oder Untersuchungsbudgets zu verbrauchen. Die konfigurierbare +`min_free_bytes`-Reserve ersetzt keine Speicherüberwachung oder Retention. Die +20-GB-Vorgabe für Entwicklungs-/Testdaten bleibt davon unabhängig. Der Worker +löscht keine Paketgenerationen und keine Inhalte erhaltener Akten. + +Ctrl+C bzw. SIGTERM fordert einen geordneten Stopp an. Der laufende Dispatch darf +noch abschließen; erst danach wird die Prozess-Lease freigegeben. Ein Dispatch +kann mehrere interne Schritte/Modellanfragen enthalten und entsprechend dauern. +Bei hartem Beenden greift beim nächsten Start der Checkpoint-Wiederanlauf. + +Für einen Server ist die Vorlage `deploy/viano-bug-analysis-worker.service` +vorbereitet. Sie wird nicht automatisch installiert oder gestartet. Dienstbenutzer, +Pfade, Zugangsdaten und gemeinsame Workspace-Berechtigungen müssen zum Zielserver +passen. Der Worker startet weder die lokale Testdatenbank noch einen Repository- +Synchronisationsdienst. Diese Abhängigkeiten werden separat betrieben. + +## Lokale Tests + +Worker-, Übergabe- und Wiederanlauftests verwenden künstliche Daten und lokale +Provideradapter; es werden keine kostenpflichtigen Modellanfragen ausgeführt: + +```powershell +.\.venv\Scripts\python.exe -m pytest tests/test_worker.py tests/profiles/vcore/test_worker_intake.py tests/profiles/vcore/test_intake_api.py -q +``` diff --git a/docs/intake-api.md b/docs/intake-api.md index 17bdd05..e9231aa 100644 --- a/docs/intake-api.md +++ b/docs/intake-api.md @@ -3,8 +3,8 @@ 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. +dauerhaft laufender [Analyseworker](analysis-worker.md) übernimmt anschließend die +Aufnahme und Verarbeitung. Die allgemeine Weboberfläche bleibt ein weiterer Ausbauschritt. ## Lokaler Start diff --git a/docs/test-results/analysis-worker-20261021.md b/docs/test-results/analysis-worker-20261021.md new file mode 100644 index 0000000..48910f0 --- /dev/null +++ b/docs/test-results/analysis-worker-20261021.md @@ -0,0 +1,56 @@ +# Lokale Prüfung des dauerhaften Analyseworkers – 2026-10-21 + +Ausgangspunkt: `27feda6` (Übergabe-API). Keine echten Provideranfragen, keine +kostenpflichtigen LLM-Tests und kein Start gegen produktive Aufträge. + +## Umfang und Ergebnis + +252 Tests bestanden, fünf bestehende Trialfälle und ein bestehender +Telemetrie-Integrationstest separat ausgeklammert. +Geprüft wurden Worker, API-Intake, Vcore-Import, Checkpoints, Queue, Casebuilder, +Paketfabrik und Paketwriter sowie die übrigen 15 Trialtests und 27 Telemetrietests. + +Die gezielten Diensttests umfassen: + +- Übernahme einer API-Einlieferung und tatsächlichen Casebau ohne manuellen Trial; +- Interpretation mit lokalem Fixture-Provider und Wiederanlauf aus einem neuen Workerobjekt; +- Wiedereröffnung einer Akte: alte Interpretation bleibt erhalten, neue Revision wird neu interpretiert; +- Projekttrennung, Fortschritt anderer Fälle trotz fehlerhaftem Fall; +- persistente Wartezeiten, unveränderte LLM-Versuchskennung; +- Bewahrung menschlicher/technischer BLOCKED-Sperren und ausgeschöpfter Modellversuche; +- explizit zugelassene technische Wiederaufnahme; +- Dienst- und Fall-Leases, geordneten Stopp, Konfigurationswechsel und Speicherreserve; +- Attention-Markierung der tatsächlich neuesten Generation bei Exception nach einem Checkpoint; +- keine Wiederveröffentlichung bei unverändertem abgeschlossenem Scheduler-Ergebnis; +- CLI-Prüfung/Status und leeren Einmalbetrieb ohne API-Clienttoken. + +Die vorhandenen fünf Trialfehler sind `INVESTIGATION_PROMPT_LIMIT` in alten +synthetischen Untersuchungsfixtures. Gegenprobe: Der vollständige unveränderte +`trial.py`-Dispatcher aus `27feda6` wurde ausschließlich im Arbeitsspeicher geladen; +dieselben fünf Tests scheiterten, 15 bestanden. Keine Quelldatei wurde dafür +zurückgesetzt. Die 20.000-Zeichen-Grenze und die Produktionsprompts wurden nicht +verändert, um diese alten Tests passend zu machen. + +Betroffene Tests: + +- `test_trial_full_investigation_delivery_and_resume` (beide Parameterfälle) +- `test_cli_run_connects_telemetry_and_publishes` +- `test_human_test_returns_blocked_with_pending_budget` +- `test_dispatch_yield_resumes_without_repeating_first_model_call` + +Der zusätzliche bestehende Test +`test_full_pipeline_observed_without_telemetry_artifacts_or_double_accounting` +scheitert in seiner alten synthetischen Pipeline vor dem Codingauftrag mit +`CHANGE_REQUEST_DIAGNOSIS_NOT_READY`. Eine Gegenprobe mit der unveränderten +Telemetrie-Allowlist aus `27feda6` scheiterte ebenfalls. Dieser Test verwendet +keinen Worker; seine Pipeline-/Prompt-Fixtures wurden hier nicht umgebaut. + +Ein Deployment als Betriebssystemdienst und echte Provider-/DB-Erreichbarkeit +wurden nicht getestet. Eine einzelne externe Modellanfrage kann bei hartem +Prozessabbruch vor dem Antwort-Checkpoint erneut erforderlich werden; die lokalen +Tests belegen keine Exactly-once-Garantie des Modellanbieters. + +Vor den Tests geprüft: Test-Workspaces 16.939.129.428 Bytes, kuratiertes Fehlerkorpus +2.471.041 Bytes und vorhandene pytest-Temporärkopien 6.143.826 Bytes. Damit bestand +Reserve unter dem Entwicklungsbudget von 20.000.000.000 Bytes. Die neuen Tests +verwenden kleine künstliche Archive. diff --git a/src/viano_bug_analysis/api/config.py b/src/viano_bug_analysis/api/config.py index eaf2312..eaf9d69 100644 --- a/src/viano_bug_analysis/api/config.py +++ b/src/viano_bug_analysis/api/config.py @@ -51,12 +51,13 @@ class Project: profile: str adapter: str = 'vcore' settings: dict = field(default_factory=dict) + planning_mode: str = 'joint' 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): + or set(self.settings) - SETTINGS or self.planning_mode not in {'joint', 'separate'}): raise APIError('API_PROFILE_CONFIG', 500) diff --git a/src/viano_bug_analysis/api/service.py b/src/viano_bug_analysis/api/service.py index 1d6f20f..b082f2f 100644 --- a/src/viano_bug_analysis/api/service.py +++ b/src/viano_bug_analysis/api/service.py @@ -57,12 +57,28 @@ class IntakeService: def status(self, submission, client): receipt = self._receipt(submission, client) status = receipt['status'] + revision = receipt['analysis_revision'] + generation = None 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', + revision = reader.manifest['analysis_revision'] + generation = reader.manifest['package_generation'] + result = {k: receipt[k] for k in ('submission_id', 'project_id', 'profile_id', 'case_id', 'package_id', 'analysis_revision')} | {'status': status} + result['analysis_revision'] = revision + if generation is not None: + result['package_generation'] = generation + ledger = self.workspace.runtime / 'worker/cases' / (key(receipt['case_id'] + ':' + receipt['package_id']) + '.json') + if ledger.is_file(): + try: + record = json.loads(ledger.read_bytes()) + if record.get('analysis_revision') == revision: + result['worker'] = {k: record.get(k) for k in ('status', 'code', 'retry_not_before', 'updated_at')} + except (OSError, ValueError, TypeError): + result['worker'] = {'status': 'UNKNOWN'} + return result def package(self, submission, client): receipt = self._receipt(submission, client) diff --git a/src/viano_bug_analysis/telemetry/codes.py b/src/viano_bug_analysis/telemetry/codes.py index b7993c9..ded8549 100644 --- a/src/viano_bug_analysis/telemetry/codes.py +++ b/src/viano_bug_analysis/telemetry/codes.py @@ -46,3 +46,16 @@ KNOWN_MODULES = KNOWN_MODULES | frozenset({"viano_bug_analysis.analysis.investig KNOWN_CODES = KNOWN_CODES | frozenset({"INVESTIGATION_ENTITY_REFERENCE"}) KNOWN_MODULES = KNOWN_MODULES | frozenset({"viano_bug_analysis.analysis.investigation_contract"}) + +# Operational service boundary added after the original analysis allowlist. +KNOWN_CODES |= frozenset({ + 'WORKER_CONFIG', 'WORKER_LLM_CONFIG_REQUIRED', 'WORKER_PROJECT_BINDING', + 'WORKER_CONFIG_CHANGED', 'WORKER_STORAGE_IDENTITY', 'WORKER_CASE_ERROR', + 'WORKER_PROJECT_UNKNOWN', 'WORKER_STORAGE_LOW', 'WORKER_CYCLE_ERROR', + 'WORKER_BUSY', 'WORKER_RETRY_STATE', 'WORKER_NOT_FOUND', +}) +KNOWN_MODULES |= frozenset({ + 'viano_bug_analysis.workflow', 'viano_bug_analysis.worker.config', + 'viano_bug_analysis.worker.service', 'viano_bug_analysis.worker.__main__', +}) +KNOWN_CLASSES |= frozenset({'WorkerError'}) diff --git a/src/viano_bug_analysis/trial.py b/src/viano_bug_analysis/trial.py index a051128..b7f07f5 100644 --- a/src/viano_bug_analysis/trial.py +++ b/src/viano_bug_analysis/trial.py @@ -25,131 +25,7 @@ from .telemetry.safety import safe_code from .analysis.correction import CorrectionConfig -class TrialError(Exception): - def __init__(self, code): - self.code = code - super().__init__(code) - - -def preflight(path): - """Validate transport configuration and credentials without sending requests.""" - config = GatewayConfig.load(path) - gateway = LLMGateway(config) - eligible = [p for p in config.providers - if p.enabled and p.privacy in PRIVACY[config.privacy_profile]] - if not eligible: - raise TrialError("LLM_PRIVACY_NO_PROVIDER") - for provider in eligible: - if provider.api_key_env: - value = os.environ.get(provider.api_key_env) - if not value or any(ord(c) < 32 or ord(c) == 127 for c in value): - raise TrialError("LLM_CREDENTIALS") - return gateway - - -def describe(path, status, *, code=None, exported=None, retry_not_before=None): - with PackageReader(path) as reader: - m = reader.manifest - return {"status": status, "code": code, "package": str(path), - "case_id": m["case_id"], "package_id": m["package_id"], - "package_generation": m["package_generation"], - "analysis_revision": m["analysis_revision"], "state": m["state"], - "routing": m["routing"], "exported": str(exported) if exported else None, - "retry_not_before": retry_not_before} - - -async def drive(store, gateway, case_id, package_id, *, max_dispatches=100, - attempt_id="default", policies=None, progress=None, stop_after=None): - if type(max_dispatches) is not int or max_dispatches <= 0: - raise TrialError("TRIAL_DISPATCH_LIMIT") - if stop_after not in {None, "case", "interpretation"}: - raise TrialError("TRIAL_STOP_PHASE") - policies = policies or {} - identity = (case_id, package_id) - correction = policies.get("correction", CorrectionConfig()) - from .tools import InvestigationToolRouter - project_research = policies.get("project_research") - interpreter = CaseInterpreter(store, gateway, correction_config=correction, project_research=project_research, - joint_config=policies.get("analysis", HypothesisConfig()) if policies.get("planning_mode") == "joint" else None) - hypotheses = HypothesisBuilder(store, gateway, policies.get("analysis"), correction_config=correction) - loop = Investigator(store, gateway, policies.get("investigation"), - budget_config=policies.get("budget"), - router=InvestigationToolRouter(project_research=project_research)) - diagnosis = DiagnosisBuilder(store, gateway, policies.get("diagnosis")) - request = ChangeRequestBuilder(store, gateway, policies.get("change_request")) - output = OutputValidator(store, gateway) - for dispatch in range(max_dispatches): - latest = store.latest(*identity) - if latest is None: - raise TrialError("CASE_NOT_FOUND") - with PackageReader(latest, store.limits) as reader: - state = reader.manifest["state"] - artifacts = reader.manifest["artifacts"] - routing = reader.manifest["routing"] - if state == State.BLOCKED: - return describe(latest, "BLOCKED", code=routing["reason"], - retry_not_before=routing["retry_not_before"]) - if phase_complete(latest, stop_after): - return describe(latest, "PHASE_COMPLETE") - if progress: - progress({"dispatch": dispatch + 1, "state": state}) - if state in {State.OUTPUT_READY, State.FAILED_TO_ANALYZE, State.HANDED_OFF}: - result = await output.publish(*identity) - elif state == State.BUILDING_CASE and artifacts.get("case"): - result = await interpreter.interpret(*identity, attempt_id=attempt_id) - elif state in {State.VALIDATED, State.REOPENED, State.BUILDING_CASE}: - result = CaseBuilder(store, project_research=project_research).build(*identity) - elif state == State.INTERPRETING_CASE: - if artifacts.get("case_interpretation"): - result = await hypotheses.build(*identity, attempt_id=attempt_id) - else: - result = await interpreter.interpret(*identity, attempt_id=attempt_id) - elif state == State.GENERATING_HYPOTHESES: - if artifacts.get("hypotheses"): - result = await loop.step(*identity) - else: - result = await hypotheses.build(*identity, attempt_id=attempt_id) - elif state == State.INVESTIGATING_HYPOTHESES: - result = await loop.step(*identity) - elif state == State.BUILDING_DIAGNOSIS: - if not artifacts.get("diagnosis"): - result = await diagnosis.build(*identity) - elif not artifacts.get("change_request"): - result = await request.build(*identity) - else: - result = await output.publish(*identity) - else: - raise TrialError("TRIAL_PHASE") - status = getattr(result, "status", "COMPLETED") - if status == "INCONCLUSIVE": - result = await output.publish(*identity) - status = result.status - if status in {"BLOCKED", "RETRY_PENDING", "DEFERRED", "ERROR", "BUSY", - "OUTPUT_READY", "REVIEW", "HANDED_OFF"}: - return describe(result.package, status, code=getattr(result, "code", None), - exported=getattr(result, "exported", None), - retry_not_before=getattr(result, "retry_not_before", None)) - latest = store.latest(*identity) - return describe(latest, "PHASE_COMPLETE" if phase_complete(latest, stop_after) else "YIELDED") - - -def phase_complete(path, stop_after): - if stop_after is None: - return False - from .case.contracts import read_json - with PackageReader(path) as reader: - manifest = reader.manifest - if stop_after == "case": - ref = manifest["artifacts"].get("case") - if manifest["state"] != State.BUILDING_CASE or not ref: - return False - doc = read_json(reader, ref, "case", 8 * 1024 * 1024)["case"] - else: - ref = manifest["artifacts"].get("case_interpretation") - if manifest["state"] != State.INTERPRETING_CASE or not ref: - return False - doc = read_json(reader, ref, "case-interpretation", 2 * 1024 * 1024) - return all(doc[field] == manifest[field] for field in ("case_id", "package_id", "analysis_revision")) +from .workflow import TrialError, preflight, describe, drive, phase_complete def positive(text): diff --git a/src/viano_bug_analysis/worker/__init__.py b/src/viano_bug_analysis/worker/__init__.py new file mode 100644 index 0000000..526df7d --- /dev/null +++ b/src/viano_bug_analysis/worker/__init__.py @@ -0,0 +1,5 @@ +"""Single-process, resumable analysis service. Importing never starts analysis.""" +from .config import WorkerConfig, WorkerError +from .service import AnalysisWorker + +__all__ = ['WorkerConfig', 'WorkerError', 'AnalysisWorker'] diff --git a/src/viano_bug_analysis/worker/__main__.py b/src/viano_bug_analysis/worker/__main__.py new file mode 100644 index 0000000..7854bac --- /dev/null +++ b/src/viano_bug_analysis/worker/__main__.py @@ -0,0 +1,72 @@ +import argparse +import asyncio +import json +import signal + +from viano_bug_analysis.api.config import directory +from viano_bug_analysis.lifecycle.storage import Workspace +from viano_bug_analysis.lifecycle.errors import LockBusy +from viano_bug_analysis.telemetry import Telemetry, TelemetryConfig +from .config import WorkerConfig, code, load_project +from .service import AnalysisWorker + + +def main(argv=None): + parser = argparse.ArgumentParser(description='Resumable BugAnalysis service worker') + parser.add_argument('command', choices=['check', 'run', 'status', 'retry']) + parser.add_argument('--config', required=True) + parser.add_argument('--once', action='store_true', help='one scheduler pass; may make paid model calls') + parser.add_argument('--case-id') + parser.add_argument('--package-id') + args = parser.parse_args(argv) + if args.command == 'retry' and not (args.case_id and args.package_id): + parser.error('retry requires --case-id and --package-id') + try: + config = WorkerConfig.load(args.config) + if args.command == 'check': + for project in config.intake.projects: + load_project(project) + if config.telemetry_config: + TelemetryConfig.load(config.telemetry_config) + result = {'status': 'CONFIG_READY', 'network_tested': False, + 'projects': [p.id for p in config.intake.projects]} + elif args.command == 'status': + path = config.intake.workspace / 'runtime/worker/status.json' + result = json.loads(path.read_bytes()) if path.is_file() else {'status': 'NOT_STARTED'} + if path.is_file(): + result['note'] = 'Last snapshot; process liveness is not guaranteed.' + else: + directory(config.intake.workspace) + workspace = Workspace(config.intake.workspace) + if args.command == 'retry': + AnalysisWorker(config).retry(args.case_id, args.package_id) + result = {'status': 'RELEASED', 'case_id': args.case_id, 'package_id': args.package_id} + else: + telemetry_config = TelemetryConfig.load(config.telemetry_config) if config.telemetry_config else None + with Telemetry(workspace, telemetry_config) as telemetry: + worker = AnalysisWorker(config, telemetry=telemetry) + previous = {} + for signum in (signal.SIGINT, signal.SIGTERM): + previous[signum] = signal.getsignal(signum) + signal.signal(signum, lambda *_: worker.request_stop()) + try: + asyncio.run(worker.run(once=args.once)) + finally: + for signum, handler in previous.items(): + signal.signal(signum, handler) + result = {'status': 'STOPPED'} + print(json.dumps(result, ensure_ascii=True)) + return 0 + except LockBusy: + print(json.dumps({'status': 'BUSY', 'code': 'WORKER_BUSY'})) + return 3 + except KeyboardInterrupt: + print(json.dumps({'status': 'INTERRUPTED'})) + return 130 + except Exception as error: + print(json.dumps({'status': 'ERROR', 'code': code(getattr(error, 'code', None))})) + return 2 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/viano_bug_analysis/worker/config.py b/src/viano_bug_analysis/worker/config.py new file mode 100644 index 0000000..4d634f3 --- /dev/null +++ b/src/viano_bug_analysis/worker/config.py @@ -0,0 +1,86 @@ +import hashlib +import tomllib +from dataclasses import dataclass +from pathlib import Path + +from viano_bug_analysis.api.config import APIConfig, APIError +from viano_bug_analysis.lifecycle.storage import json_bytes +from viano_bug_analysis.telemetry.safety import safe_code + +WORKER_CODES = frozenset({'WORKER_CONFIG', 'WORKER_LLM_CONFIG_REQUIRED', 'WORKER_PROJECT_BINDING', + 'WORKER_CONFIG_CHANGED', 'WORKER_STORAGE_IDENTITY', 'WORKER_CASE_ERROR', 'WORKER_PROJECT_UNKNOWN', + 'WORKER_STORAGE_LOW', 'WORKER_CYCLE_ERROR', 'WORKER_BUSY', 'WORKER_RETRY_STATE', 'WORKER_NOT_FOUND'}) + + +def code(value): + return value if isinstance(value, str) and value in WORKER_CODES else safe_code(value) + + +class WorkerError(Exception): + def __init__(self, code): + self.code = code + super().__init__(code) + + +@dataclass(frozen=True) +class WorkerConfig: + intake: APIConfig + poll_seconds: int = 5 + error_retry_seconds: int = 60 + max_inbox_per_pass: int = 20 + min_free_bytes: int = 512 * 1024**2 + telemetry_config: Path | None = None + + def __post_init__(self): + for name in ('poll_seconds', 'error_retry_seconds', 'max_inbox_per_pass', 'min_free_bytes'): + if type(getattr(self, name)) is not int or getattr(self, name) <= 0: + raise WorkerError('WORKER_CONFIG') + + @classmethod + def load(cls, path): + try: + path = Path(path).resolve() + data = tomllib.loads(path.read_text(encoding='utf-8')) + if set(data) != {'worker'}: + raise ValueError() + row = dict(data['worker']) + intake = APIConfig.load(path.parent / row.pop('intake_config')) + if row.get('telemetry_config'): + row['telemetry_config'] = (path.parent / row['telemetry_config']).resolve() + return cls(intake=intake, **row) + except (OSError, ValueError, TypeError, KeyError, APIError): + raise WorkerError('WORKER_CONFIG') from None + + +def binding(project): + """No secrets/config contents enter the runtime ledger, only file digests.""" + files = {name: hashlib.sha256(Path(path).read_bytes()).hexdigest() + for name, path in project.settings.items()} + return hashlib.sha256(json_bytes({'project': project.id, 'type': project.type, + 'profile': project.profile, 'planning_mode': project.planning_mode, 'files': files})).hexdigest() + + +def load_project(project): + from viano_bug_analysis.workflow import preflight + from viano_bug_analysis.analysis import (HypothesisConfig, InvestigationConfig, + DiagnosisConfig, ChangeRequestConfig, CorrectionConfig) + from viano_bug_analysis.budget import BudgetConfig + from viano_bug_analysis.tools.project_query import ProjectResearch + if 'llm_config' not in project.settings: + raise WorkerError('WORKER_LLM_CONFIG_REQUIRED') + before = binding(project) + gateway = preflight(project.settings['llm_config']) + policies = {'planning_mode': project.planning_mode} + for name, cls in (('analysis', HypothesisConfig), ('investigation', InvestigationConfig), + ('diagnosis', DiagnosisConfig), ('change_request', ChangeRequestConfig), + ('correction', CorrectionConfig), ('budget', BudgetConfig)): + if path := project.settings.get(name + '_config'): + policies[name] = cls.load(path) + if path := project.settings.get('project_config'): + research = ProjectResearch.load(path) + if research.project.get('type') != project.type or research.project.get('profile') != project.profile: + raise WorkerError('WORKER_PROJECT_BINDING') + policies['project_research'] = research + if before != binding(project): + raise WorkerError('WORKER_CONFIG_CHANGED') + return gateway, policies, before diff --git a/src/viano_bug_analysis/worker/service.py b/src/viano_bug_analysis/worker/service.py new file mode 100644 index 0000000..94ea74a --- /dev/null +++ b/src/viano_bug_analysis/worker/service.py @@ -0,0 +1,287 @@ +import asyncio +import json +import os +import shutil +from datetime import datetime, timedelta, timezone + +from viano_bug_analysis.api.config import directory +from viano_bug_analysis.api.service import IntakeService +from viano_bug_analysis.lifecycle import InboxProcessor, State +from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain +from viano_bug_analysis.lifecycle.errors import LockBusy +from viano_bug_analysis.lifecycle.job import timestamp +from viano_bug_analysis.lifecycle.locking import FileLock +from viano_bug_analysis.lifecycle.storage import atomic_json, key +from viano_bug_analysis.package import PackageReader +from viano_bug_analysis.workflow import drive +from .config import WorkerError, binding, load_project, code as safe_code + +class AnalysisWorker: + """One analysis step per eligible case/pass; case files remain authoritative.""" + def __init__(self, config, *, telemetry=None, loader=load_project, dispatcher=drive, clock=None): + self.config, self.loader, self.dispatcher = config, loader, dispatcher + self.clock = clock or (lambda: datetime.now(timezone.utc)) + self.intake = IntakeService(config.intake) + self.workspace = self.intake.workspace + self.queue = InboxProcessor(self.workspace, telemetry=telemetry) + self.store = self.queue.store + self.telemetry = telemetry + self.root = directory(self.workspace.runtime / 'worker') + self.records = directory(self.root / 'cases') + self.projects = {p.id: p for p in config.intake.projects} + self.runtimes = {} + self.stop_requested = False + self.last_case = None + self.last_inbox = None + + def request_stop(self): + self.stop_requested = True + + def _record_path(self, manifest): + return self.records / (key(manifest['case_id'] + ':' + manifest['package_id']) + '.json') + + def _record(self, manifest): + path = self._record_path(manifest) + return json.loads(path.read_bytes()) if path.exists() else {} + + def _save(self, manifest, status, *, code=None, due=None, config_hash=None): + previous = self._record(manifest) + record = {name: manifest[name] for name in ('case_id', 'package_id', 'package_generation', 'analysis_revision')} + record.update(status=status, code=safe_code(code) if code else None, + retry_not_before=due, updated_at=timestamp(), + config_hash=config_hash or (previous.get('config_hash') if + previous.get('analysis_revision') == manifest['analysis_revision'] else None)) + atomic_json(self._record_path(manifest), record) + + def _status(self, status, *, code=None, case=None): + record = {'status': status, 'updated_at': timestamp(), 'pid': os.getpid(), + 'code': safe_code(code) if code else None, 'case': case} + atomic_json(self.root / 'status.json', record) + + def _due(self, value): + return value is None or datetime.fromisoformat(value.replace('Z', '+00:00')) <= self.clock() + + def _pause(self, manifest, code, config_hash=None, status='ERROR'): + due = (self.clock() + timedelta(seconds=self.config.error_retry_seconds)).isoformat() + self._save(manifest, status, code=code, due=due, config_hash=config_hash) + + def _heads(self): + # Inspect only the latest candidate, never reopen every historical ZIP. + # Filenames select candidates; the manifest and CheckpointStore validate identity. + for parent in sorted(self.workspace.cases.iterdir()): + if not parent.is_dir(): + continue + try: + directory(parent) + except Exception: + self._status('DEGRADED', code='WORKER_STORAGE_IDENTITY') + continue + for folder in sorted(parent.iterdir()): + if not folder.is_dir(): + continue + try: + directory(folder) + except Exception: + self._status('DEGRADED', code='WORKER_STORAGE_IDENTITY') + continue + paths = sorted(folder.glob('g*.zip')) + if not paths: + continue + try: + if paths[-1].is_symlink() or getattr(paths[-1].lstat(), 'st_file_attributes', 0) & 0x400: + raise WorkerError('WORKER_STORAGE_IDENTITY') + with PackageReader(paths[-1], self.store.limits) as reader: + manifest = reader.manifest + if folder != self.workspace.case_directory(manifest['case_id'], manifest['package_id']): + raise WorkerError('WORKER_STORAGE_IDENTITY') + latest = self.store.latest(manifest['case_id'], manifest['package_id']) + with PackageReader(latest, self.store.limits) as reader: + yield reader.manifest + except Exception as error: + # A broken case does not prevent other cases from progressing. + self._status('DEGRADED', code=getattr(error, 'code', 'WORKER_CASE_ERROR')) + + def _runtime(self, project): + current = binding(project) + runtime = self.runtimes.get(project.id) + if runtime and runtime[2] != current: + # Reload configuration only after restart. Avoid hot swaps mid-operation. + raise WorkerError('WORKER_CONFIG_CHANGED') + if runtime is None: + runtime = self.loader(project) + self.runtimes[project.id] = runtime + return runtime + + def retry(self, case_id, package_id): + """Release only a scheduler error; never bypass case/operation budgets.""" + with FileLock(self.workspace.runtime / 'locks/analysis-service.lock'): + latest = self.store.latest(case_id, package_id) + if not latest: + raise WorkerError('WORKER_NOT_FOUND') + with PackageReader(latest, self.store.limits) as reader: + manifest = reader.manifest + record = self._record(manifest) + project = self.projects.get(manifest['project'].get('id')) + if not project: + raise WorkerError('WORKER_PROJECT_UNKNOWN') + if (record.get('analysis_revision') == manifest['analysis_revision'] + and record.get('config_hash') not in {None, binding(project)}): + raise WorkerError('WORKER_CONFIG_CHANGED') + if manifest['state'] == State.BLOCKED: + # Operator confirms the read-only research service is repaired. + # No human-feedback, model-validation or exhausted LLM reset. + if manifest['routing']['reason'] != 'RESEARCH_UNAVAILABLE' or not self._due(manifest['routing']['retry_not_before']): + raise WorkerError('WORKER_RETRY_STATE') + with PackageReader(latest, self.store.limits) as reader: + resume = checkpoint_chain(reader)[0][1]['resume_state'] + latest = self.store.advance(case_id, package_id, resume, + step_id=f"worker-operator-retry:{manifest['package_generation']}", + details={'operation': 'worker-operator-retry'}, + expected_generation=manifest['package_generation']) + with PackageReader(latest, self.store.limits) as reader: + manifest = reader.manifest + self._save(manifest, 'RELEASED') + + def _admit(self): + sources = sorted(self.workspace.inbox.glob('*.zip')) + if self.last_inbox: + sources.sort(key=lambda p: (p.name <= self.last_inbox, p.name)) + for source in sources[:self.config.max_inbox_per_pass]: + if self.stop_requested: + break + self.last_inbox = source.name + try: + # Share the case lease with interactive trials, also during intake. + try: + with PackageReader(source, self.store.limits) as reader: + case_id = reader.manifest['case_id'] + except Exception: + self.queue.process(source) # normal validator/quarantine behavior + continue + with FileLock(self.workspace.runtime / 'locks' / ('trial-' + key(case_id) + '.lock')): + result = self.queue.process(source) + if result.status in {'ERROR', 'BUSY'}: + self._status('DEGRADED', code=result.code) + except LockBusy: + continue + + async def tick(self): + self._status('RUNNING') + if shutil.disk_usage(self.workspace.root).free < self.config.min_free_bytes: + self._status('STORAGE_LOW', code='WORKER_STORAGE_LOW') + return 0 + self._admit() + heads = list(self._heads()) + heads.sort(key=lambda m: (m['case_id'], m['package_id'])) + # Rotate between passes so a repeatedly busy case cannot monopolize first place. + if self.last_case: + heads.sort(key=lambda m: ((m['case_id'], m['package_id']) <= self.last_case, + m['case_id'], m['package_id'])) + count = 0 + for manifest in heads: + if self.stop_requested: + break + identity = manifest['case_id'], manifest['package_id'] + dispatch_started = False + try: + with FileLock(self.workspace.runtime / 'locks' / ('trial-' + key(identity[0]) + '.lock')): + latest = self.store.latest(*identity) + with PackageReader(latest, self.store.limits) as reader: + manifest = reader.manifest + chain = checkpoint_chain(reader) + record = self._record(manifest) + same_revision = record.get('analysis_revision') == manifest['analysis_revision'] + same_generation = record.get('package_generation') == manifest['package_generation'] + if same_generation and record.get('status') in {'DELIVERED', 'ATTENTION'}: + continue + if same_generation and not self._due(record.get('retry_not_before')): + continue + project = self.projects.get(manifest['project'].get('id')) + if not project or (project.type, project.profile) != ( + manifest['project'].get('type'), manifest['project'].get('profile')): + self._pause(manifest, 'WORKER_PROJECT_UNKNOWN', status='ATTENTION') + continue + gateway, policies, config_hash = self._runtime(project) + if same_revision and record.get('config_hash') not in {None, config_hash}: + self._pause(manifest, 'WORKER_CONFIG_CHANGED', status='ATTENTION') + continue + if manifest['state'] == State.BLOCKED: + cp = chain[0][1] + if (manifest['routing']['reason'] == 'AWAITING_TEST_FEEDBACK' + or cp['details'].get('automatic_retry') is False + or manifest['routing']['reason'] in {'INVALID_LLM_RESPONSES', 'LLM_CONFIGURATION'}): + self._save(manifest, 'BLOCKED', code=manifest['routing']['reason'], config_hash=config_hash) + continue + if not self._due(manifest['routing']['retry_not_before']): + continue + # An exhausted LLM operation cannot be made runnable by + # changing its attempt id. It requires operator intervention. + if manifest['routing']['reason'] == 'LLM_UNAVAILABLE': + self._save(manifest, 'ATTENTION', code='LLM_RETRY_EXHAUSTED', config_hash=config_hash) + continue + latest = self.store.advance(*identity, cp['resume_state'], + step_id=f"retry:{manifest['package_generation']}", expected_generation=manifest['package_generation']) + with PackageReader(latest, self.store.limits) as reader: + manifest = reader.manifest + # Persist binding before a model call, not after it. + self._save(manifest, 'DISPATCHING', config_hash=config_hash) + self._status('ANALYZING', case={'case_id': identity[0], 'package_id': identity[1]}) + dispatch_started = True + result = await self.dispatcher(self.store, gateway, *identity, + max_dispatches=1, attempt_id='default', policies=policies) + count += 1 + self.last_case = identity + with PackageReader(self.store.latest(*identity), self.store.limits) as reader: + current = reader.manifest + status = result['status'] + if status in {'OUTPUT_READY', 'REVIEW', 'HANDED_OFF'}: + self._save(current, 'DELIVERED', config_hash=config_hash) + elif status in {'RETRY_PENDING', 'DEFERRED', 'BUSY'}: + self._pause(current, result.get('code') or status, config_hash, status=status) + if result.get('retry_not_before'): + self._save(current, status, code=result.get('code'), due=result['retry_not_before'], config_hash=config_hash) + elif status == 'ERROR': + self._save(current, 'ATTENTION', code=result.get('code'), config_hash=config_hash) + else: + self._save(current, status, code=result.get('code'), config_hash=config_hash) + except LockBusy: + continue + except Exception as error: + if self.telemetry: + self.telemetry.guard(self.telemetry.exception, error) + try: + # The phase may have committed before raising. Suppress the + # actual latest generation, not the stale input generation. + with PackageReader(self.store.latest(*identity), self.store.limits) as reader: + manifest = reader.manifest + if dispatch_started: + self._save(manifest, 'ATTENTION', code=getattr(error, 'code', 'WORKER_CASE_ERROR')) + else: + self._pause(manifest, getattr(error, 'code', 'WORKER_CASE_ERROR')) + except Exception: + self._status('DEGRADED', code='WORKER_CASE_ERROR') + self._status('IDLE') + return count + + async def run(self, *, once=False): + with FileLock(self.workspace.runtime / 'locks/analysis-service.lock'): + try: + self.intake.recover() + while not self.stop_requested: + progressed = 0 + try: + progressed = await self.tick() + except Exception as error: + self._status('DEGRADED', code=getattr(error, 'code', 'WORKER_CYCLE_ERROR')) + if once: + break + if progressed: + await asyncio.sleep(0) + continue + # Event-loop sleep keeps signal handling responsive. + for _ in range(self.config.poll_seconds * 10): + if self.stop_requested: + break + await asyncio.sleep(0.1) + finally: + self._status('STOPPED') diff --git a/src/viano_bug_analysis/workflow.py b/src/viano_bug_analysis/workflow.py new file mode 100644 index 0000000..da94325 --- /dev/null +++ b/src/viano_bug_analysis/workflow.py @@ -0,0 +1,155 @@ +"""Shared resumable analysis dispatch used by trials and the service worker.""" +import os +from .analysis import (CaseInterpreter, HypothesisBuilder, HypothesisConfig, Investigator, + DiagnosisBuilder, ChangeRequestBuilder, OutputValidator) +from .analysis.correction import CorrectionConfig +from .case import CaseBuilder +from .lifecycle import State +from .llm import GatewayConfig, LLMGateway +from .llm.gateway import PRIVACY +from .package import PackageReader + +class TrialError(Exception): + def __init__(self, code): + self.code = code + super().__init__(code) + + +def preflight(path): + """Validate transport configuration and credentials without sending requests.""" + config = GatewayConfig.load(path) + gateway = LLMGateway(config) + eligible = [p for p in config.providers + if p.enabled and p.privacy in PRIVACY[config.privacy_profile]] + if not eligible: + raise TrialError("LLM_PRIVACY_NO_PROVIDER") + for provider in eligible: + if provider.api_key_env: + value = os.environ.get(provider.api_key_env) + if not value or any(ord(c) < 32 or ord(c) == 127 for c in value): + raise TrialError("LLM_CREDENTIALS") + return gateway + + +def describe(path, status, *, code=None, exported=None, retry_not_before=None): + with PackageReader(path) as reader: + m = reader.manifest + return {"status": status, "code": code, "package": str(path), + "case_id": m["case_id"], "package_id": m["package_id"], + "package_generation": m["package_generation"], + "analysis_revision": m["analysis_revision"], "state": m["state"], + "routing": m["routing"], "exported": str(exported) if exported else None, + "retry_not_before": retry_not_before} + + +async def drive(store, gateway, case_id, package_id, *, max_dispatches=100, + attempt_id="default", policies=None, progress=None, stop_after=None): + if type(max_dispatches) is not int or max_dispatches <= 0: + raise TrialError("TRIAL_DISPATCH_LIMIT") + if stop_after not in {None, "case", "interpretation"}: + raise TrialError("TRIAL_STOP_PHASE") + policies = policies or {} + identity = (case_id, package_id) + correction = policies.get("correction", CorrectionConfig()) + from .tools import InvestigationToolRouter + project_research = policies.get("project_research") + interpreter = CaseInterpreter(store, gateway, correction_config=correction, project_research=project_research, + joint_config=policies.get("analysis", HypothesisConfig()) if policies.get("planning_mode") == "joint" else None) + hypotheses = HypothesisBuilder(store, gateway, policies.get("analysis"), correction_config=correction) + loop = Investigator(store, gateway, policies.get("investigation"), + budget_config=policies.get("budget"), + router=InvestigationToolRouter(project_research=project_research)) + diagnosis = DiagnosisBuilder(store, gateway, policies.get("diagnosis")) + request = ChangeRequestBuilder(store, gateway, policies.get("change_request")) + output = OutputValidator(store, gateway) + for dispatch in range(max_dispatches): + latest = store.latest(*identity) + if latest is None: + raise TrialError("CASE_NOT_FOUND") + with PackageReader(latest, store.limits) as reader: + state = reader.manifest["state"] + artifacts = dict(reader.manifest["artifacts"]) + routing = reader.manifest["routing"] + # A reopened case retains prior artifacts. They are history, not + # completed stages of the new revision. + revision = reader.manifest['analysis_revision'] + from .lifecycle.checkpoints import checkpoint_chain + chain = checkpoint_chain(reader) + for pointer, step in ( + ('case', f'case-builder:v1:revision:{revision}:complete'), + ('case_interpretation', f'case-interpreter:v1:revision:{revision}:complete'), + ('hypotheses', f'hypothesis-builder:v1:revision:{revision}:complete'), + ('diagnosis', f'diagnosis:v1:{revision}:complete'), + ('change_request', f'change-request:v1:{revision}:complete')): + if artifacts.get(pointer) and not any(cp['step_id'] == step for _, cp in chain): + # Contracts identify current products even when newer phase + # implementations use a different completion step name. + import json + doc = json.loads(reader.read_bytes(artifacts[pointer])) + artifact_identity = doc.get('case', doc) + if artifact_identity.get('analysis_revision') != revision: + artifacts[pointer] = None + if state == State.BLOCKED: + return describe(latest, "BLOCKED", code=routing["reason"], + retry_not_before=routing["retry_not_before"]) + if phase_complete(latest, stop_after): + return describe(latest, "PHASE_COMPLETE") + if progress: + progress({"dispatch": dispatch + 1, "state": state}) + if state in {State.OUTPUT_READY, State.FAILED_TO_ANALYZE, State.HANDED_OFF}: + result = await output.publish(*identity) + elif state == State.BUILDING_CASE and artifacts.get("case"): + result = await interpreter.interpret(*identity, attempt_id=attempt_id) + elif state in {State.VALIDATED, State.REOPENED, State.BUILDING_CASE}: + result = CaseBuilder(store, project_research=project_research).build(*identity) + elif state == State.INTERPRETING_CASE: + if artifacts.get("case_interpretation"): + result = await hypotheses.build(*identity, attempt_id=attempt_id) + else: + result = await interpreter.interpret(*identity, attempt_id=attempt_id) + elif state == State.GENERATING_HYPOTHESES: + if artifacts.get("hypotheses"): + result = await loop.step(*identity) + else: + result = await hypotheses.build(*identity, attempt_id=attempt_id) + elif state == State.INVESTIGATING_HYPOTHESES: + result = await loop.step(*identity) + elif state == State.BUILDING_DIAGNOSIS: + if not artifacts.get("diagnosis"): + result = await diagnosis.build(*identity) + elif not artifacts.get("change_request"): + result = await request.build(*identity) + else: + result = await output.publish(*identity) + else: + raise TrialError("TRIAL_PHASE") + status = getattr(result, "status", "COMPLETED") + if status == "INCONCLUSIVE": + result = await output.publish(*identity) + status = result.status + if status in {"BLOCKED", "RETRY_PENDING", "DEFERRED", "ERROR", "BUSY", + "OUTPUT_READY", "REVIEW", "HANDED_OFF"}: + return describe(result.package, status, code=getattr(result, "code", None), + exported=getattr(result, "exported", None), + retry_not_before=getattr(result, "retry_not_before", None)) + latest = store.latest(*identity) + return describe(latest, "PHASE_COMPLETE" if phase_complete(latest, stop_after) else "YIELDED") + + +def phase_complete(path, stop_after): + if stop_after is None: + return False + from .case.contracts import read_json + with PackageReader(path) as reader: + manifest = reader.manifest + if stop_after == "case": + ref = manifest["artifacts"].get("case") + if manifest["state"] != State.BUILDING_CASE or not ref: + return False + doc = read_json(reader, ref, "case", 8 * 1024 * 1024)["case"] + else: + ref = manifest["artifacts"].get("case_interpretation") + if manifest["state"] != State.INTERPRETING_CASE or not ref: + return False + doc = read_json(reader, ref, "case-interpretation", 2 * 1024 * 1024) + return all(doc[field] == manifest[field] for field in ("case_id", "package_id", "analysis_revision")) diff --git a/tests/profiles/vcore/test_worker_intake.py b/tests/profiles/vcore/test_worker_intake.py new file mode 100644 index 0000000..6f8b010 --- /dev/null +++ b/tests/profiles/vcore/test_worker_intake.py @@ -0,0 +1,30 @@ +import asyncio +import json + +from viano_bug_analysis.api import APIConfig, Client, Project, IntakeService, IntakeLimits +from viano_bug_analysis.package import PackageReader +from viano_bug_analysis.worker import AnalysisWorker, WorkerConfig +from viano_bug_analysis.worker.config import binding +from test_case_interpreter import gateway, FixtureAdapter + + +def test_api_submission_is_admitted_and_built_without_manual_trial(source_pair, tmp_path): + client = Client('capture', 'UNUSED_API_TOKEN', ('core',)) + project = Project('core', 'wow', 'vcore') + config = APIConfig(tmp_path / 'service', (project,), (client,), limits=IntakeLimits(min_free_bytes=1)) + service = IntakeService(config) + data = {'kind': 'new_case', 'project_id': 'core', 'profile_id': 'vcore', 'case_id': 'BUG-000020'} + receipt, _ = service.submit(client, 'capture1', json.dumps(data).encode(), [('wowdiag', source_pair[0])]) + def load(p): + return gateway(FixtureAdapter()), {}, binding(p) + worker = AnalysisWorker(WorkerConfig(config, min_free_bytes=1), loader=load) + asyncio.run(worker.run(once=True)) + assert not list(worker.workspace.inbox.glob('*.zip')) + latest = worker.store.latest('BUG-000020', receipt['package_id']) + with PackageReader(latest) as reader: + assert reader.manifest['state'] == 'BUILDING_CASE' + assert reader.manifest['artifacts']['case'] + status = service.status(receipt['submission_id'], client) + assert status['status'] == 'BUILDING_CASE' + assert status['analysis_revision'] == 1 + assert status['worker']['status'] == 'YIELDED' diff --git a/tests/test_worker.py b/tests/test_worker.py new file mode 100644 index 0000000..e2049d0 --- /dev/null +++ b/tests/test_worker.py @@ -0,0 +1,283 @@ +import asyncio +import json +from dataclasses import replace +from datetime import datetime, timedelta, timezone + +import pytest + +from viano_bug_analysis.api import APIConfig, Client, Project +from viano_bug_analysis.lifecycle import State +from viano_bug_analysis.lifecycle.locking import FileLock +from viano_bug_analysis.lifecycle.errors import LockBusy +from viano_bug_analysis.package import PackageReader +from viano_bug_analysis.worker import AnalysisWorker, WorkerConfig, WorkerError +from viano_bug_analysis.worker.config import binding +from test_case_builder import case_setup +from test_case_interpreter import gateway, FixtureAdapter + +CASE = ('case-A', 'stable-package') + + +@pytest.fixture(autouse=True) +def project_manifest(manifest): + manifest['project'] = {'id': 'test', 'type': 'wow', 'profile': 'vcore'} + + +@pytest.fixture +def setup(case_setup): + store = case_setup[0] + project = Project('test', 'wow', 'vcore', planning_mode='separate') + config = WorkerConfig(APIConfig(store.workspace.root, (project,), + (Client('capture', 'TEST_TOKEN', ('test',)),)), min_free_bytes=1) + def loader(p): + return gateway(FixtureAdapter()), {'planning_mode': 'separate'}, binding(p) + return config, loader, store + + +def test_real_case_build_then_interpret_and_restart(setup): + config, loader, store = setup + worker = AnalysisWorker(config, loader=loader) + assert asyncio.run(worker.tick()) == 1 + with PackageReader(store.latest(*CASE)) as reader: + assert reader.manifest['state'] == State.BUILDING_CASE + assert reader.manifest['artifacts']['case'] + # Fresh worker object has no in-memory queue. It recovers from managed packages. + resumed = AnalysisWorker(config, loader=loader) + assert asyncio.run(resumed.tick()) == 1 + with PackageReader(store.latest(*CASE)) as reader: + assert reader.manifest['state'] == State.INTERPRETING_CASE + assert reader.manifest['artifacts']['case_interpretation'] + + +def test_scheduler_retry_is_durable_and_not_busy_loop(setup): + config, loader, store = setup + now = datetime.now(timezone.utc) + calls = [] + async def dispatch(*args, **kwargs): + calls.append(kwargs) + return {'status': 'RETRY_PENDING', 'code': 'LLM_CONNECTION', + 'retry_not_before': (now + timedelta(minutes=5)).isoformat()} + worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now) + assert asyncio.run(worker.tick()) == 1 + again = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now) + assert asyncio.run(again.tick()) == 0 + assert len(calls) == 1 and calls[0]['attempt_id'] == 'default' + after = AnalysisWorker(config, loader=loader, dispatcher=dispatch, + clock=lambda: now + timedelta(minutes=6)) + assert asyncio.run(after.tick()) == 1 + + +@pytest.mark.parametrize('reason,automatic', [ + ('AWAITING_TEST_FEEDBACK', True), ('INVALID_LLM_RESPONSES', True), + ('LLM_UNAVAILABLE', True), ('RESEARCH_UNAVAILABLE', False)]) +def test_worker_never_silently_bypasses_block(setup, reason, automatic): + config, loader, store = setup + store.advance(*CASE, State.BUILDING_CASE, step_id='start') + blocked = store.advance(*CASE, State.BLOCKED, step_id='block', reason=reason, + retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': automatic}) + async def forbidden(*args, **kwargs): + raise AssertionError('blocked case was dispatched') + worker = AnalysisWorker(config, loader=loader, dispatcher=forbidden) + assert asyncio.run(worker.tick()) == 0 + assert store.latest(*CASE) == blocked + if reason != 'RESEARCH_UNAVAILABLE': + with pytest.raises(WorkerError, match='WORKER_RETRY_STATE'): + worker.retry(*CASE) + else: + worker.retry(*CASE) + with PackageReader(store.latest(*CASE)) as reader: + assert reader.manifest['state'] == 'BUILDING_CASE' + + +def test_explicitly_permitted_due_retry_resumes(setup): + config, loader, store = setup + store.advance(*CASE, State.BUILDING_CASE, step_id='start') + store.advance(*CASE, State.BLOCKED, step_id='block', reason='RESEARCH_UNAVAILABLE', + retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': True}) + seen = [] + async def dispatch(store, gateway, *identity, **kwargs): + with PackageReader(store.latest(*identity)) as reader: + seen.append(reader.manifest['state']) + return {'status': 'YIELDED'} + assert asyncio.run(AnalysisWorker(config, loader=loader, dispatcher=dispatch).tick()) == 1 + assert seen == ['BUILDING_CASE'] + + +def test_one_service_and_trial_lease_exclusion(setup): + config, loader, store = setup + worker = AnalysisWorker(config, loader=loader) + with FileLock(worker.workspace.runtime / 'locks/analysis-service.lock'): + with pytest.raises(LockBusy): + asyncio.run(worker.run(once=True)) + from viano_bug_analysis.lifecycle.storage import key + with FileLock(worker.workspace.runtime / 'locks' / ('trial-' + key(CASE[0]) + '.lock')): + assert asyncio.run(worker.tick()) == 0 + + +def test_exception_backoff_and_explicit_scheduler_release(setup): + config, loader, store = setup + calls = [] + async def broken(*args, **kwargs): + calls.append(1) + raise ValueError('secret data must not appear in runtime state') + worker = AnalysisWorker(config, loader=loader, dispatcher=broken) + asyncio.run(worker.tick()) + asyncio.run(worker.tick()) + assert len(calls) == 1 + record = next(worker.records.glob('*.json')).read_text() + assert 'secret data' not in record + worker.retry(*CASE) + asyncio.run(worker.tick()) + assert len(calls) == 2 + + +def test_stop_requested_during_dispatch_finishes_step(setup): + config, loader, store = setup + async def dispatch(*args, **kwargs): + worker.request_stop() + return {'status': 'YIELDED'} + worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch) + asyncio.run(worker.run()) + assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STOPPED' + assert next(worker.records.glob('*.json')).exists() + + +def test_config_binding_persisted_before_dispatch_and_detects_restart_change(setup): + config, loader, store = setup + async def dispatch(*args, **kwargs): + assert next(worker.records.glob('*.json')).exists() + return {'status': 'YIELDED'} + worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch) + asyncio.run(worker.tick()) + changed = replace(config.intake.projects[0], planning_mode='joint') + updated = replace(config, intake=replace(config.intake, projects=(changed,))) + other = AnalysisWorker(updated, loader=loader, dispatcher=dispatch) + assert asyncio.run(other.tick()) == 0 + record = json.loads(next(other.records.glob('*.json')).read_bytes()) + assert record['status'] == 'ATTENTION' and record['code'] == 'WORKER_CONFIG_CHANGED' + + +def test_unknown_project_held_without_model_call(setup): + config, loader, store = setup + changed = replace(config.intake.projects[0], id='other') + updated = replace(config, intake=APIConfig(config.intake.workspace, (changed,), + (Client('capture', 'TEST_TOKEN', ('other',)),))) + def forbidden(project): + raise AssertionError('unknown project') + worker = AnalysisWorker(updated, loader=forbidden) + assert asyncio.run(worker.tick()) == 0 + assert json.loads(next(worker.records.glob('*.json')).read_bytes())['code'] == 'WORKER_PROJECT_UNKNOWN' + + +def test_terminal_result_not_republished_every_pass(setup): + config, loader, store = setup + calls = [] + async def dispatch(*args, **kwargs): + calls.append(1) + return {'status': 'OUTPUT_READY'} + worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch) + asyncio.run(worker.tick()) + asyncio.run(worker.tick()) + assert len(calls) == 1 + + +def test_reopened_revision_does_not_skip_new_interpretation(setup): + config, loader, store = setup + worker = AnalysisWorker(config, loader=loader) + asyncio.run(worker.tick()) + asyncio.run(worker.tick()) + for state in ('GENERATING_HYPOTHESES', 'INVESTIGATING_HYPOTHESES', 'BUILDING_DIAGNOSIS', 'OUTPUT_READY', 'HANDED_OFF'): + store.advance(*CASE, state, step_id='test:' + state) + store.advance(*CASE, 'REOPENED', step_id='test:reopen') + asyncio.run(worker.tick()) + with PackageReader(store.latest(*CASE)) as reader: + assert reader.manifest['analysis_revision'] == 2 + assert reader.manifest['state'] == 'BUILDING_CASE' + # Old interpretation is deliberately retained in the append-only package. + old = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation'])) + assert old['analysis_revision'] == 1 + asyncio.run(worker.tick()) + with PackageReader(store.latest(*CASE)) as reader: + doc = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation'])) + assert doc['analysis_revision'] == 2 + + +def test_multiple_projects_and_poison_case_do_not_starve_other_cases(setup, case_setup, tmp_path): + from viano_bug_analysis.package.factory import PackageFactory + config, loader, store = setup + with PackageReader(case_setup[2]) as reader: + manifest = reader.manifest + contents = {ref: reader.read_bytes(ref) for ref in reader.files if ref != 'manifest.json'} + manifest.update(case_id='case-B', package_id='other-package', project={'id': 'second', 'type': 'wow', 'profile': 'vcore'}) + PackageFactory().create(store.workspace.inbox / 'other.zip', manifest, contents) + project2 = Project('second', 'wow', 'vcore', planning_mode='separate') + intake = replace(config.intake, projects=config.intake.projects + (project2,)) + seen = [] + def load(project): + return project.id, {'project_id': project.id}, binding(project) + async def dispatch(store, gateway, case_id, package_id, **kwargs): + seen.append((case_id, gateway, kwargs['policies']['project_id'])) + if case_id == 'case-A': + raise ValueError('broken first case') + return {'status': 'YIELDED'} + worker = AnalysisWorker(replace(config, intake=intake), loader=load, dispatcher=dispatch) + asyncio.run(worker.tick()) + assert seen == [('case-A', 'test', 'test'), ('case-B', 'second', 'second')] + asyncio.run(worker.tick()) + assert [v[0] for v in seen].count('case-A') == 1 + assert [v[0] for v in seen].count('case-B') == 2 + + +def test_cli_check_status_and_empty_once_do_not_need_api_token(tmp_path, monkeypatch, capsys): + from viano_bug_analysis.worker.__main__ import main + import viano_bug_analysis.worker.__main__ as cli + monkeypatch.delenv('ABSENT_API_TOKEN', raising=False) + intake = tmp_path / 'api.toml' + intake.write_text('''[api] +workspace = "service" +[[projects]] +id = "test" +type = "wow" +profile = "vcore" +[[clients]] +id = "capture" +token_env = "ABSENT_API_TOKEN" +projects = ["test"] +''') + config = tmp_path / 'worker.toml' + config.write_text('[worker]\nintake_config = "api.toml"\nmin_free_bytes = 1\n') + loaded = [] + monkeypatch.setattr(cli, 'load_project', lambda p: loaded.append(p.id)) + assert main(['check', '--config', str(config)]) == 0 + assert loaded == ['test'] + assert main(['run', '--once', '--config', str(config)]) == 0 + assert main(['status', '--config', str(config)]) == 0 + assert 'STOPPED' in capsys.readouterr().out + + +def test_storage_floor_pauses_without_dispatch(setup, monkeypatch): + import viano_bug_analysis.worker.service as module + from collections import namedtuple + config, loader, store = setup + Usage = namedtuple('Usage', 'total used free') + worker = AnalysisWorker(config, loader=loader) + monkeypatch.setattr(module.shutil, 'disk_usage', lambda *_: Usage(100, 100, 0)) + assert asyncio.run(worker.tick()) == 0 + assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STORAGE_LOW' + + +def test_exception_after_checkpoint_holds_actual_latest_generation(setup): + config, loader, store = setup + calls = [] + async def dispatch(store, gateway, *identity, **kwargs): + calls.append(1) + store.advance(*identity, 'BUILDING_CASE', step_id='commit-before-error') + raise ValueError('unexpected after checkpoint') + worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch) + asyncio.run(worker.tick()) + asyncio.run(worker.tick()) + assert len(calls) == 1 + record = json.loads(next(worker.records.glob('*.json')).read_bytes()) + with PackageReader(store.latest(*CASE)) as reader: + assert record['package_generation'] == reader.manifest['package_generation'] + assert record['status'] == 'ATTENTION'