from viano_bug_analysis.telemetry.decorators import observed import hashlib from dataclasses import dataclass from pathlib import Path from viano_bug_analysis.case.contracts import read_json, require_file, validate from viano_bug_analysis.case.errors import CaseBuildError from viano_bug_analysis.lifecycle import LifecycleError, State from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain from viano_bug_analysis.lifecycle.errors import LockBusy from viano_bug_analysis.lifecycle.locking import FileLock from viano_bug_analysis.lifecycle.storage import json_bytes, key from viano_bug_analysis.llm import CaseLLMRunner, LLMError from viano_bug_analysis.package import PackageReader, PackageReadError from viano_bug_analysis.package.manifest import parse_manifest from viano_bug_analysis.package.validator import ARCHIVE_ERRORS from .contracts import OBSERVATIONS from .errors import InterpretationError from .hypothesis_config import HypothesisConfig, HypothesisError, HypothesisLimits from .hypothesis_contracts import build_context, records_for, registry_for, validate_generation, validate_proposals from .hypothesis_prompt import HypothesisPromptBuilder from .hypothesis_summary import render_hypotheses from .interpreter import CaseInterpreter @dataclass(frozen=True) class HypothesisBuildResult: status: str package: Path analysis_revision: int hypotheses_ref: str | None = None registry_ref: str | None = None summary_ref: str | None = None reused: bool = False retry_not_before: str | None = None code: str | None = None class HypothesisBuilder: def __init__(self, store, gateway, config=None, limits=None, *, interpretation_limits=None, case_limits=None, clock=None, correction_config=None): self.store = store self.gateway = gateway self.config = config or HypothesisConfig() self.limits = limits or HypothesisLimits() self.interpreter = CaseInterpreter(store, gateway, interpretation_limits, case_limits=case_limits, clock=clock) self.runner = CaseLLMRunner(store, gateway, clock=clock) self.prompt = HypothesisPromptBuilder() self.correction_config = correction_config _correction_result = staticmethod(HypothesisBuildResult) def _request(self, context): request = self.prompt.build(context) from .correction import augment, current_attempt attempt = current_attempt(self) if attempt: identity = context["case_context"]["case"] with PackageReader(self.store.latest(identity["case_id"], identity["package_id"]), self.store.limits) as reader: request = augment(request, reader, "hypotheses", attempt) return request @observed("hypotheses_build") async def build(self, case_id, package_id, *, attempt_id="default"): if not isinstance(attempt_id, str) or not attempt_id.strip(): raise HypothesisError("HYPOTHESIS_ATTEMPT") try: with FileLock(self.store.workspace.runtime / "locks" / f"hypotheses-{key(case_id)}.lock"): from .correction import dispatch return await dispatch(self, "hypotheses", case_id, package_id, attempt_id, self._build) except LockBusy: raise except (CaseBuildError, InterpretationError, LifecycleError, LLMError, PackageReadError) as error: raise HypothesisError(getattr(error, "code", "HYPOTHESIS_PACKAGE")) from None except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, UnicodeError, RecursionError): raise HypothesisError("HYPOTHESIS_INPUT") from None async def _build(self, case_id, package_id, attempt_id): latest = self.store.latest(case_id, package_id) if latest is None: raise HypothesisError("HYPOTHESIS_INTERPRETATION_NOT_READY") with PackageReader(latest, self.store.limits) as reader: revision = reader.manifest["analysis_revision"] interpretation_id = f"case-interpreter:v1:revision:{revision}:complete" if not any(cp["step_id"] == interpretation_id for _, cp in checkpoint_chain(reader)): raise HypothesisError("HYPOTHESIS_INTERPRETATION_NOT_READY") # Reuse only a confirmed interpretation; never initiate that earlier LLM phase. confirmed = await self.interpreter.interpret(case_id, package_id) with PackageReader(confirmed.package, self.store.limits) as reader: manifest = reader.manifest interpretation = read_json(reader, confirmed.interpretation_ref, "case-interpretation", self.interpreter.limits.max_output_bytes) case_context = read_json(reader, interpretation["context_ref"], "case-interpretation-context", self.interpreter.limits.max_context_bytes) chain = checkpoint_chain(reader) base = f"analysis/hypotheses/{revision:04d}" complete_id = f"hypothesis-builder:v1:revision:{revision}:complete" start_id = f"hypothesis-builder:v1:revision:{revision}:start" completed = next((cp for _, cp in chain if cp["step_id"] == complete_id), None) started = next((cp for _, cp in chain if cp["step_id"] == start_id), None) policy_ref, context_ref = base + "/policy.json", base + "/context.json" policy = read_json(reader, policy_ref, "hypothesis-policy", self.limits.max_context_bytes) if started else self.config.policy() prior = self._history(reader, chain, revision) context = build_context(case_context, interpretation, confirmed.interpretation_ref, policy, prior, self.limits) if started: saved_context = read_json(reader, context_ref, "hypothesis-context", self.limits.max_context_bytes) version = saved_context.get('prompt_version', 'hypotheses_v1') else: version = 'hypotheses_v4' if version in {'hypotheses_v2','hypotheses_v3','hypotheses_v4'}: context['prompt_version'] = version if len(json_bytes(context)) > self.limits.max_context_bytes: raise HypothesisError('HYPOTHESIS_CONTEXT_LIMIT') elif version != 'hypotheses_v1': raise HypothesisError('HYPOTHESIS_CONTEXT_CHANGED') if completed: return self._reuse(reader, context, completed, complete_id) self._guard_existing_registry(reader, chain, revision) if manifest["state"] == State.BLOCKED: return HypothesisBuildResult("BLOCKED", confirmed.package, revision, retry_not_before=manifest["routing"]["retry_not_before"], code=manifest["routing"]["reason"] or "CASE_BLOCKED") if manifest["state"] not in {State.INTERPRETING_CASE, State.GENERATING_HYPOTHESES}: raise HypothesisError("HYPOTHESIS_PHASE") if not any(entry["kind"] in OBSERVATIONS for entry in case_context["sources"].values()) or not interpretation["interpretation"]["observed_behavior"]: raise HypothesisError("HYPOTHESIS_NO_OBSERVATIONS") if started and policy != self.config.policy(): raise HypothesisError("HYPOTHESIS_POLICY_CHANGED") encoded_context, encoded_policy = json_bytes(context), json_bytes(policy) context_hash, policy_hash = hashlib.sha256(encoded_context).hexdigest(), hashlib.sha256(encoded_policy).hexdigest() start_details = {"operation": "hypotheses-generating", "builder_version": "1", "context_ref": context_ref, "context_sha256": context_hash, "policy_ref": policy_ref, "policy_sha256": policy_hash, "interpretation_ref": confirmed.interpretation_ref, "interpretation_sha256": context["interpretation_sha256"]} request = self.gateway.validate_request(self._request(context)) if started: stored_context = read_json(reader, context_ref, "hypothesis-context", self.limits.max_context_bytes) if started["details"] != start_details or stored_context != context: raise HypothesisError("HYPOTHESIS_CONTEXT_CHANGED") else: self.store.advance(case_id, package_id, State.GENERATING_HYPOTHESES, step_id=start_id, details=start_details, additions={context_ref: encoded_context, policy_ref: encoded_policy}, artifacts={"hypothesis_context": context_ref, "hypothesis_policy": policy_ref, "hypotheses": None, "hypotheses_summary": None, "hypothesis_registry": None}, expected_generation=manifest["package_generation"]) operation_id = f"hypotheses:v1:{attempt_id}" outcome = await self.runner.run(case_id, package_id, operation_id=operation_id, request=request) if outcome.status not in {"SUCCESS", "ERROR"}: return HypothesisBuildResult(outcome.status, outcome.package, revision, retry_not_before=outcome.retry_not_before, code=outcome.code) with PackageReader(outcome.package, self.store.limits) as reader: current = reader.manifest if (current["state"] != State.GENERATING_HYPOTHESES or current["analysis_revision"] != revision or current["artifacts"]["case_interpretation"] != confirmed.interpretation_ref): raise HypothesisError("HYPOTHESIS_PHASE") call = read_json(reader, outcome.result_ref, "llm-call", self.gateway.config.max_response_bytes + self.gateway.config.max_request_bytes + 16384) if (call["status"] != outcome.status or call["phase"] != State.GENERATING_HYPOTHESES or any(call[field] != current[field] for field in ("case_id", "package_id", "analysis_revision")) or call["input_request_hash"] != hashlib.sha256(json_bytes(request)).hexdigest() or call["operation_key"] != key(f"{revision}:{operation_id}")): raise HypothesisError("HYPOTHESIS_LLM_BINDING") common = {"format_version": 1, "builder_version": "1", "prompt_version": version, "case_id": case_id, "package_id": package_id, "analysis_revision": revision, "case_ref": interpretation["case_ref"], "case_sha256": interpretation["case_sha256"], "interpretation_ref": confirmed.interpretation_ref, "interpretation_sha256": context["interpretation_sha256"], "context_ref": context_ref, "context_sha256": context_hash, "policy_ref": policy_ref, "policy_sha256": policy_hash, "llm_result_ref": outcome.result_ref} if outcome.status == "ERROR": return self._reject(reader, common, attempt_id, outcome.code) hypotheses_ref, registry_ref, summary_ref = base + "/hypotheses.json", base + "/registry.json", base + "/hypotheses.md" try: payload = call["outcome"]["document"] validate_proposals(payload, context) document = {**common, "format": "viano-hypotheses", "registry_ref": registry_ref, "policy": policy, "budget_accounting": "not_started", "llm_metadata": call["outcome"]["metadata"], "hypotheses": records_for(payload, policy)} registry = registry_for(document, hypotheses_ref) validate_generation(document, registry, context, hypotheses_ref) encoded, registry_bytes = json_bytes(document), json_bytes(registry) summary = render_hypotheses(document).encode("utf-8") if max(len(encoded), len(registry_bytes)) > self.limits.max_output_bytes: raise HypothesisError("HYPOTHESIS_OUTPUT_LIMIT") if len(summary) > self.limits.max_summary_bytes: raise HypothesisError("HYPOTHESIS_SUMMARY_LIMIT") except HypothesisError as error: return self._reject(reader, common, attempt_id, error.code) details = {"operation": "hypotheses-built", "builder_version": "1", "context_ref": context_ref, "context_sha256": context_hash, "policy_ref": policy_ref, "policy_sha256": policy_hash, "hypotheses_ref": hypotheses_ref, "registry_ref": registry_ref, "summary_ref": summary_ref, "llm_result_ref": outcome.result_ref, "created_count": registry["created_count"]} artifacts = {"hypotheses": hypotheses_ref, "hypotheses_summary": summary_ref, "hypothesis_registry": registry_ref, "hypothesis_context": context_ref, "hypothesis_policy": policy_ref} result = self.store.advance(case_id, package_id, State.GENERATING_HYPOTHESES, step_id=complete_id, details=details, additions={hypotheses_ref: encoded, registry_ref: registry_bytes, summary_ref: summary}, artifacts=artifacts, expected_generation=current["package_generation"]) return HypothesisBuildResult("COMPLETED", result, revision, hypotheses_ref, registry_ref, summary_ref) def _history(self, reader, chain, revision): prior = [] for _, checkpoint in reversed(chain): if checkpoint["analysis_revision"] >= revision or checkpoint["details"].get("operation") != "hypotheses-built": continue if len(prior) >= self.limits.max_history_documents: raise HypothesisError("HYPOTHESIS_HISTORY_LIMIT") ref = checkpoint["details"]["hypotheses_ref"] document = read_json(reader, ref, "hypotheses", self.limits.max_output_bytes) if any(document[field] != reader.manifest[field] for field in ("case_id", "package_id")) or document["analysis_revision"] != checkpoint["analysis_revision"]: raise HypothesisError("HYPOTHESIS_HISTORY_IDENTITY") prior.append({"analysis_revision": document["analysis_revision"], "hypotheses_ref": ref, "hypotheses_sha256": hashlib.sha256(json_bytes(document)).hexdigest(), "initial_proposals": [record["proposal"] for record in document["hypotheses"]]}) return prior def _guard_existing_registry(self, reader, chain, revision): if any(cp["analysis_revision"] == revision and cp["details"].get("operation") in {"hypotheses-built", "hypothesis-created"} for _, cp in chain): raise HypothesisError("HYPOTHESIS_EXISTING_REGISTRY") for field, contract in (("hypothesis_registry", "hypothesis-registry"), ("hypotheses", "hypotheses")): ref = reader.manifest["artifacts"].get(field) if ref: existing = read_json(reader, ref, contract, self.limits.max_output_bytes) if any(existing[name] != reader.manifest[name] for name in ("case_id", "package_id")) or existing["analysis_revision"] >= revision: raise HypothesisError("HYPOTHESIS_EXISTING_REGISTRY") def _reject(self, reader, common, attempt_id, code): document = {**common, "format": "viano-hypothesis-error", "code": code} validate(document, "hypothesis-error") error_ref = f"analysis/hypotheses/{common['analysis_revision']:04d}/errors/{key(attempt_id)}.json" step_id = f"hypothesis-builder:v1:revision:{common['analysis_revision']}:attempt:{key(attempt_id)}:rejected" reused = any(cp["step_id"] == step_id for _, cp in checkpoint_chain(reader)) details = {"operation": "hypotheses-rejected", "builder_version": "1", "error_ref": error_ref, "context_ref": common["context_ref"], "policy_ref": common["policy_ref"], "llm_result_ref": common["llm_result_ref"], "code": code} result = self.store.advance(common["case_id"], common["package_id"], State.GENERATING_HYPOTHESES, step_id=step_id, details=details, additions={error_ref: json_bytes(document)}, expected_generation=reader.manifest["package_generation"]) self.runner.logger.error("Hypothesis generation rejected case_id=%s code=%s", common["case_id"], code) return HypothesisBuildResult("ERROR", result, common["analysis_revision"], reused=reused, code=code) def _reuse(self, reader, context, checkpoint, step_id): details = checkpoint["details"] document = read_json(reader, details["hypotheses_ref"], "hypotheses", self.limits.max_output_bytes) registry = read_json(reader, details["registry_ref"], "hypothesis-registry", self.limits.max_output_bytes) stored_context = read_json(reader, details["context_ref"], "hypothesis-context", self.limits.max_context_bytes) policy = read_json(reader, details["policy_ref"], "hypothesis-policy", self.limits.max_context_bytes) if stored_context != context or document["policy"] != policy or context["policy"] != policy: raise HypothesisError("HYPOTHESIS_CONTEXT_CHANGED") validate_generation(document, registry, context, details["hypotheses_ref"]) if any(document[field] != reader.manifest[field] for field in ("case_id", "package_id", "analysis_revision")): raise HypothesisError("HYPOTHESIS_IDENTITY") if (document["context_sha256"] != hashlib.sha256(json_bytes(context)).hexdigest() or document["policy_sha256"] != hashlib.sha256(json_bytes(policy)).hexdigest() or document["interpretation_sha256"] != context["interpretation_sha256"]): raise HypothesisError("HYPOTHESIS_CONTEXT_CHANGED") expected_fields = {"context_ref": details["context_ref"], "context_sha256": document["context_sha256"], "policy_ref": details["policy_ref"], "policy_sha256": document["policy_sha256"], "registry_ref": details["registry_ref"], "llm_result_ref": details["llm_result_ref"]} if any(document[field] != value for field, value in expected_fields.items()): raise HypothesisError("HYPOTHESIS_LLM_BINDING") if (document["case_ref"] != context["case_context"]["case_ref"] or document["case_sha256"] != context["case_context"]["case_sha256"] or document["interpretation_ref"] != context["interpretation_ref"] or details["created_count"] != registry["created_count"] or details["context_sha256"] != document["context_sha256"] or details["policy_sha256"] != document["policy_sha256"]): raise HypothesisError("HYPOTHESIS_LLM_BINDING") call = read_json(reader, document["llm_result_ref"], "llm-call", self.gateway.config.max_response_bytes + self.gateway.config.max_request_bytes + 16384) payload = {"hypotheses": [record["proposal"] for record in document["hypotheses"]]} if (call["status"] != "SUCCESS" or call["phase"] != State.GENERATING_HYPOTHESES or any(call[field] != document[field] for field in ("case_id", "package_id", "analysis_revision")) or call["outcome"].get("document") != payload or call["outcome"].get("metadata") != document["llm_metadata"]): raise HypothesisError("HYPOTHESIS_LLM_BINDING") request_ref = document["llm_result_ref"].rsplit("/", 1)[0] + "/request.json" require_file(reader, request_ref) with reader._archive.open(request_ref) as stream: raw = reader.validator._read_bounded(stream, min(self.gateway.config.max_request_bytes, self.store.limits.max_file_bytes)) saved_request = parse_manifest(raw) if not isinstance(saved_request, dict): raise HypothesisError("HYPOTHESIS_LLM_BINDING") expected_request = self.prompt.build(context) from .correction import replay_request expected_request = replay_request(reader, expected_request, call, "hypotheses") expected_request["privacy"] = saved_request.get("privacy") if saved_request != expected_request or call["input_request_hash"] != hashlib.sha256(raw).hexdigest(): raise HypothesisError("HYPOTHESIS_LLM_BINDING") summary = render_hypotheses(document).encode("utf-8") if len(summary) > self.limits.max_summary_bytes: raise HypothesisError("HYPOTHESIS_SUMMARY_LIMIT") require_file(reader, details["summary_ref"]) with reader._archive.open(details["summary_ref"]) as stream: stored_summary = reader.validator._read_bounded(stream, min(self.limits.max_summary_bytes, self.store.limits.max_file_bytes)) artifacts = {"hypotheses": details["hypotheses_ref"], "hypotheses_summary": details["summary_ref"], "hypothesis_registry": details["registry_ref"], "hypothesis_context": details["context_ref"], "hypothesis_policy": details["policy_ref"]} if stored_summary != summary or any(reader.manifest["artifacts"].get(field) != ref for field, ref in artifacts.items()): raise HypothesisError("HYPOTHESIS_OUTPUT_CHANGED") latest = self.store.advance(document["case_id"], document["package_id"], State.GENERATING_HYPOTHESES, step_id=step_id, details=details, additions={details["hypotheses_ref"]: json_bytes(document), details["registry_ref"]: json_bytes(registry), details["summary_ref"]: summary}, artifacts=artifacts) if latest != reader.path: raise HypothesisError("HYPOTHESIS_GENERATION_CHANGED") return HypothesisBuildResult("COMPLETED", latest, document["analysis_revision"], details["hypotheses_ref"], details["registry_ref"], details["summary_ref"], True)