Add persistent profile-routed analysis worker and shared workflow

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).

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"

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"

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

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
+```

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
 

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.

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)
 
 

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)

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'})

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):

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']

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())

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

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')

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"))

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'

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'