460 lines
31 KiB
Python
460 lines
31 KiB
Python
"""Durable final evaluation; deliberately does not create a coding task."""
|
|
from viano_bug_analysis.telemetry.decorators import observed
|
|
import json
|
|
import hashlib
|
|
import tomllib
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from importlib.resources import files
|
|
from pathlib import Path
|
|
|
|
from viano_bug_analysis.budget import BudgetManager, BudgetError
|
|
from viano_bug_analysis.case import CaseAssembler
|
|
from viano_bug_analysis.case.contracts import read_json, validate
|
|
from viano_bug_analysis.case.errors import CaseBuildError
|
|
from viano_bug_analysis.evidence import EvidenceStore, EvidenceError
|
|
from viano_bug_analysis.evidence.store import digest
|
|
from viano_bug_analysis.lifecycle import State, LifecycleError
|
|
from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain
|
|
from viano_bug_analysis.lifecycle.locking import FileLock
|
|
from viano_bug_analysis.lifecycle.errors import LockBusy
|
|
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 .investigator import Investigator, PROVIDER_CONFIGURATION_ERRORS
|
|
from .investigation_config import InvestigationError
|
|
|
|
|
|
class DiagnosisError(Exception):
|
|
def __init__(self, code):
|
|
self.code = code
|
|
super().__init__(code)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DiagnosisConfig:
|
|
max_invalid_responses: int = 3
|
|
|
|
def __post_init__(self):
|
|
if type(self.max_invalid_responses) is not int or self.max_invalid_responses <= 0:
|
|
raise DiagnosisError("DIAGNOSIS_CONFIG")
|
|
|
|
@classmethod
|
|
def load(cls, path):
|
|
try:
|
|
with open(path, "rb") as stream:
|
|
config = tomllib.load(stream)
|
|
if set(config) != {"diagnosis"} or set(config["diagnosis"]) - {"max_invalid_responses"}:
|
|
raise DiagnosisError("DIAGNOSIS_CONFIG")
|
|
return cls(**config["diagnosis"])
|
|
except (OSError, TypeError, ValueError):
|
|
raise DiagnosisError("DIAGNOSIS_CONFIG") from None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DiagnosisLimits:
|
|
"""Adjustable safety defaults, not package format constraints."""
|
|
max_context_bytes: int = 2 * 1024**2
|
|
max_json_bytes: int = 4 * 1024**2
|
|
|
|
def __post_init__(self):
|
|
if any(type(v) is not int or v <= 0 for v in asdict(self).values()):
|
|
raise DiagnosisError("DIAGNOSIS_LIMIT_CONFIG")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DiagnosisResult:
|
|
status: str
|
|
package: Path
|
|
diagnosis_ref: str | None = None
|
|
failure_ref: str | None = None
|
|
reused: bool = False
|
|
code: str | None = None
|
|
retry_not_before: str | None = None
|
|
|
|
|
|
def build_prompt(context, *, modern=None, feedback=None):
|
|
modern = context.get('evaluation_protocol') == 2 if modern is None else modern
|
|
schema = json.loads(files("viano_bug_analysis.schemas").joinpath("diagnosis-output-v2.schema.json" if modern else "diagnosis-output-v1.schema.json").read_text())
|
|
version="diagnosis_early_v1" if context["investigation"].get("early_stop",{}).get("status")=="requested" else "diagnosis_v1"
|
|
if modern:version="diagnosis_early_v2" if version=="diagnosis_early_v1" else "diagnosis_v2"
|
|
visible={**context, **({"validation_feedback":feedback} if feedback else {})}
|
|
return {"prompt_version": version, "privacy": "configured", "output_schema": schema,
|
|
"messages": [{"role": "system", "content": files("viano_bug_analysis.llm.prompts").joinpath(version+".txt").read_text(encoding="utf-8")},
|
|
{"role": "user", "content": json_bytes({"task": "final_diagnosis", "input": visible, "output_schema": schema}).decode()}]}
|
|
|
|
|
|
def validate_evaluation(payload, context):
|
|
if context.get('evaluation_protocol')==2 and 'decision_basis' not in payload:
|
|
raise DiagnosisError('DIAGNOSIS_DECISION_BASIS')
|
|
if 'decision_basis' in payload:
|
|
from jsonschema import Draft202012Validator
|
|
schema=json.loads(files('viano_bug_analysis.schemas').joinpath('diagnosis-output-v2.schema.json').read_text())
|
|
if not Draft202012Validator(schema).is_valid(payload):
|
|
raise DiagnosisError('DIAGNOSIS_DECISION_BASIS')
|
|
else:
|
|
validate(payload, "diagnosis-output")
|
|
hypotheses = context["investigation"]["hypotheses"]
|
|
evaluations = payload["evaluations"]
|
|
if len(evaluations) != len(hypotheses) or {e["hypothesis_id"] for e in evaluations} != set(hypotheses):
|
|
raise DiagnosisError("DIAGNOSIS_HYPOTHESES")
|
|
evidence = context["evidence"]
|
|
def check_refs(refs):
|
|
if any(ref not in evidence for ref in refs):
|
|
raise DiagnosisError("DIAGNOSIS_EVIDENCE_REFERENCE")
|
|
strict = context.get('evaluation_protocol')==2 or 'decision_basis' in payload
|
|
def positive(refs):
|
|
return any(evidence[ref]["content"]["items"]
|
|
and (not strict or evidence[ref]["content"].get("coverage")!="schema_only") for ref in refs)
|
|
for evaluation in evaluations:
|
|
check_refs(evaluation["evidence_refs"])
|
|
check_refs(evaluation["countercheck"]["evidence_refs"])
|
|
if evaluation["verdict"] == "supported" and not positive(evaluation["evidence_refs"]):
|
|
raise DiagnosisError("DIAGNOSIS_UNGROUNDED")
|
|
if evaluation["countercheck"]["adequate"] and not positive(evaluation["countercheck"]["evidence_refs"]):
|
|
raise DiagnosisError("DIAGNOSIS_COUNTERCHECK")
|
|
basis=payload.get('decision_basis')
|
|
if basis:
|
|
for contradiction in basis['contradictions']:check_refs(contradiction['evidence_refs'])
|
|
if payload['status']=='confirmed' and not basis['sufficient_for_change']:
|
|
raise DiagnosisError('DIAGNOSIS_INSUFFICIENT_CAUSAL_SUPPORT')
|
|
if payload['status']=='confirmed' and any(c['resolution']=='unresolved' for c in basis['contradictions']):
|
|
raise DiagnosisError('DIAGNOSIS_UNRESOLVED_CONTRADICTION')
|
|
if payload['status']=='confirmed' and any(e['verdict']=='unresolved' for e in evaluations) and not payload['remaining_uncertainty']:
|
|
raise DiagnosisError('DIAGNOSIS_UNCERTAINTY_REQUIRED')
|
|
if payload["status"] == "inconclusive":
|
|
if payload["selected_hypothesis"] is not None or payload["root_cause"] is not None or not payload["recommended_human_actions"]:
|
|
raise DiagnosisError("DIAGNOSIS_INCONCLUSIVE")
|
|
return
|
|
hid = payload["selected_hypothesis"]
|
|
root = payload["root_cause"]
|
|
if hid not in hypotheses or root is None or payload["confidence"] not in {"supported", "strongly_supported"}:
|
|
raise DiagnosisError("DIAGNOSIS_SELECTION")
|
|
check_refs(root["evidence_refs"])
|
|
selected = next(e for e in evaluations if e["hypothesis_id"] == hid)
|
|
if selected["verdict"] != "supported" or not positive(root["evidence_refs"]) or not set(root["evidence_refs"]).issubset(selected["evidence_refs"]):
|
|
raise DiagnosisError("DIAGNOSIS_UNGROUNDED")
|
|
if hypotheses[hid]["disposition"] == "reject":
|
|
raise DiagnosisError("DIAGNOSIS_REJECTED_SELECTION")
|
|
alternatives = [e for e in evaluations if e["hypothesis_id"] != hid]
|
|
if not any(e['countercheck']['adequate'] and positive(e['countercheck']['evidence_refs'])
|
|
for e in alternatives):
|
|
raise DiagnosisError("DIAGNOSIS_COUNTERCHECK")
|
|
|
|
|
|
def render_diagnosis(document):
|
|
# Treat model text as literal human-readable text, never embedded HTML.
|
|
import html
|
|
p = document["diagnosis"]
|
|
escape = lambda value: html.escape(str(value))
|
|
lines = ["# Final diagnosis", "", "Status: " + p["status"], "", escape(p["reason"]), ""]
|
|
if p["root_cause"]:
|
|
lines += [escape(p["root_cause"]["summary"]), "", escape(p["root_cause"]["technical_description"]), "", "Evidence: " + ", ".join(p["root_cause"]["evidence_refs"]), ""]
|
|
for e in p["evaluations"]:
|
|
lines += [f"## {e['hypothesis_id']}: {e['verdict']}", "", escape(e["reason"]), "", "Countercheck: " + escape(e["countercheck"]["reason"]), ""]
|
|
if p.get('decision_basis'):
|
|
basis=p['decision_basis']
|
|
lines += ['## Decision basis', '', escape(basis['causal_explanation']), '',
|
|
'Sufficient for bounded change: '+str(basis['sufficient_for_change']), '',
|
|
*['- '+escape(limit) for limit in basis['scope_limits']], '']
|
|
for contradiction in basis['contradictions']:
|
|
lines += [escape(contradiction['resolution'])+': '+escape(contradiction['reason']),
|
|
'Evidence: '+', '.join(contradiction['evidence_refs']), '']
|
|
lines += ["## Remaining uncertainty", "", *["- " + escape(s) for s in p["remaining_uncertainty"]], "", "## Human review actions", "", *["- " + escape(s) for s in p["recommended_human_actions"]]]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
class DiagnosisBuilder:
|
|
def __init__(self, store, gateway, config=None, limits=None, *, router=None, clock=None):
|
|
self.store, self.gateway = store, gateway
|
|
self.config = config or DiagnosisConfig()
|
|
self.config_explicit = config is not None
|
|
self.limits = limits or DiagnosisLimits()
|
|
self.loop = Investigator(store, gateway, router=router)
|
|
self.runner = CaseLLMRunner(store, gateway, clock=clock)
|
|
self.clock = self.runner.clock
|
|
|
|
@observed("diagnosis_build")
|
|
async def build(self, case_id, package_id, *, retry_id=None):
|
|
if retry_id is not None and (not isinstance(retry_id, str) or not retry_id.strip()):
|
|
raise DiagnosisError("DIAGNOSIS_RETRY_ID")
|
|
try:
|
|
with FileLock(self.store.workspace.runtime / "locks" / f"diagnosis-{key(case_id)}.lock"):
|
|
return await self._build(case_id, package_id, retry_id)
|
|
except LockBusy:
|
|
raise
|
|
except (CaseBuildError, BudgetError, EvidenceError, InvestigationError, LifecycleError, LLMError, PackageReadError) as error:
|
|
raise DiagnosisError(getattr(error, "code", "DIAGNOSIS_PACKAGE")) from None
|
|
except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, IndexError, UnicodeError, RecursionError):
|
|
raise DiagnosisError("DIAGNOSIS_INPUT") from None
|
|
|
|
def _read(self, reader, ref, contract):
|
|
return read_json(reader, ref, contract, self.limits.max_json_bytes)
|
|
|
|
async def _build(self, case_id, package_id, retry_id):
|
|
path = self.store.latest(case_id, package_id)
|
|
if path is None:
|
|
raise DiagnosisError("DIAGNOSIS_CASE_NOT_FOUND")
|
|
with PackageReader(path, self.store.limits) as reader:
|
|
manifest = reader.manifest
|
|
revision = manifest["analysis_revision"]
|
|
chain = checkpoint_chain(reader)
|
|
completed = next((cp for _, cp in chain if cp["step_id"] == f"diagnosis:v1:{revision}:complete"), None)
|
|
if manifest["state"] == State.BLOCKED:
|
|
return DiagnosisResult("BLOCKED", path, code=manifest["routing"]["reason"], retry_not_before=manifest["routing"]["retry_not_before"])
|
|
if not completed and manifest["state"] != State.BUILDING_DIAGNOSIS:
|
|
raise DiagnosisError("DIAGNOSIS_PHASE")
|
|
state = self.loop._load(reader)
|
|
if not state or not state["completed"] or state["active"]:
|
|
raise DiagnosisError("DIAGNOSIS_INVESTIGATION_NOT_READY")
|
|
case = self._read(reader, manifest["artifacts"]["case"], "case")
|
|
if CaseAssembler(self.loop.router.case_limits).assemble(reader)["inputs"]["request_hash"] != case["inputs"]["request_hash"]:
|
|
raise DiagnosisError("DIAGNOSIS_STALE_CASE")
|
|
from .early_stop import namespace, pending_accounts
|
|
prefix,base=namespace(revision,state)
|
|
early=state.get("early_stop",{}).get("status")=="requested"
|
|
started = next((cp for _, cp in chain if cp["step_id"] == prefix+":start"), None)
|
|
context_ref = base + "/context.json"
|
|
policy = self.config.max_invalid_responses
|
|
if started:
|
|
old = self._read(reader, context_ref, "diagnosis-context")
|
|
if manifest["artifacts"].get("diagnosis_context") != context_ref:
|
|
raise DiagnosisError("DIAGNOSIS_POINTER_CHANGED")
|
|
policy = old["max_invalid_responses"]
|
|
if self.config_explicit and policy != self.config.max_invalid_responses:
|
|
raise DiagnosisError("DIAGNOSIS_POLICY_CHANGED")
|
|
budget = BudgetManager(self.store).snapshot(case_id, package_id)
|
|
if (pending_accounts(budget.state) if early else not budget.ready_for_evaluation):
|
|
raise DiagnosisError("DIAGNOSIS_BUDGET_NOT_READY")
|
|
refs = EvidenceStore(self.store, router=self.loop.router).snapshot(case_id, package_id).record_refs
|
|
with PackageReader(path, self.store.limits) as reader:
|
|
evidence = {}
|
|
for ref in refs:
|
|
record = self._read(reader, ref, "evidence-record")
|
|
if record["analysis_revision"] == revision:
|
|
evidence[record["id"]] = {"record_ref": ref, "record": record, "content": self._read(reader, record["content_ref"], "evidence-content")}
|
|
interpretation = self._read(reader, manifest["artifacts"]["case_interpretation"], "case-interpretation")
|
|
context = {"format": "viano-diagnosis-context", "format_version": 1, "case": case,
|
|
"interpretation": interpretation["interpretation"], "investigation": state,
|
|
"investigation_ref": manifest["artifacts"]["investigation_state"], "budget": budget.state,
|
|
"evidence": evidence, "max_invalid_responses": policy}
|
|
if not started or old.get('evaluation_protocol')==2:
|
|
context['evaluation_protocol']=2
|
|
validate(context, "diagnosis-context")
|
|
if len(json_bytes(context)) > self.limits.max_context_bytes:
|
|
raise DiagnosisError("DIAGNOSIS_CONTEXT_LIMIT")
|
|
if started and (old != context or started["details"]["context_sha256"] != digest(context)):
|
|
raise DiagnosisError("DIAGNOSIS_CONTEXT_CHANGED")
|
|
if completed:
|
|
return self._reuse(reader, completed, context)
|
|
failures = [cp for _, cp in chain if cp["analysis_revision"] == revision and cp["details"].get("operation") == "diagnosis-rejected" and cp["step_id"].startswith(prefix+":rejected:")]
|
|
attempt = len(failures) + 1
|
|
if attempt > policy:
|
|
raise DiagnosisError("DIAGNOSIS_REJECTION_LIMIT")
|
|
# Persist the input before provider/privacy validation so configuration
|
|
# failures are blocked with the same frozen context on later resume.
|
|
request = self._request(context,reader,attempt,modern=context.get("evaluation_protocol")==2 or bool(failures))
|
|
if not started:
|
|
path = self.store.advance(case_id, package_id, State.BUILDING_DIAGNOSIS, step_id=prefix+":start",
|
|
details={"operation": "diagnosis-start", "context_ref": context_ref, "context_sha256": digest(context)},
|
|
additions={context_ref: json_bytes(context)}, artifacts={"diagnosis_context": context_ref}, expected_generation=manifest["package_generation"])
|
|
operation = f"diagnosis:v1:early:{state['early_stop']['number']}:attempt:{attempt}" if early else f"diagnosis:v1:attempt:{attempt}"
|
|
if retry_id:
|
|
# Only explicitly reopen an exhausted/configuration-failed transport.
|
|
with PackageReader(path, self.store.limits) as reader:
|
|
previous = next((cp for _, cp in checkpoint_chain(reader) if cp["details"].get("operation_key") == key(f"{revision}:{operation}")), None)
|
|
if previous and (previous["details"].get("status") == "BLOCKED" or previous["details"].get("code") in PROVIDER_CONFIGURATION_ERRORS):
|
|
operation += ":retry:" + key(retry_id)
|
|
try:
|
|
outcome = await self.runner.run(case_id, package_id, operation_id=operation, request=request)
|
|
except LLMError as error:
|
|
if error.code not in PROVIDER_CONFIGURATION_ERRORS:
|
|
raise
|
|
return self._block(case_id, package_id, revision, operation, error.code)
|
|
if outcome.status not in {"SUCCESS", "ERROR"}:
|
|
return DiagnosisResult(outcome.status, outcome.package, code=outcome.code, retry_not_before=outcome.retry_not_before)
|
|
if outcome.status == "ERROR" and outcome.code in PROVIDER_CONFIGURATION_ERRORS:
|
|
return self._block(case_id, package_id, revision, operation, outcome.code)
|
|
if BudgetManager(self.store).snapshot(case_id, package_id).state != context["budget"]:
|
|
raise DiagnosisError("DIAGNOSIS_CONTEXT_CHANGED")
|
|
with PackageReader(outcome.package, self.store.limits) as reader:
|
|
if (reader.manifest["state"] != State.BUILDING_DIAGNOSIS or reader.manifest["analysis_revision"] != revision
|
|
or self.loop._load(reader) != context["investigation"]
|
|
or CaseAssembler(self.loop.router.case_limits).assemble(reader)["inputs"]["request_hash"] != case["inputs"]["request_hash"]):
|
|
raise DiagnosisError("DIAGNOSIS_CONTEXT_CHANGED")
|
|
call = self._read_call(reader, outcome.result_ref)
|
|
self._bind(call, reader, context, outcome.result_ref, operation)
|
|
if call["status"] != outcome.status:
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
fallback = False
|
|
try:
|
|
if outcome.status == "ERROR":
|
|
raise DiagnosisError(outcome.code or "DIAGNOSIS_OUTPUT")
|
|
payload = call["outcome"]["document"]
|
|
if "decision_basis" in request["output_schema"]["required"] and "decision_basis" not in payload:
|
|
raise DiagnosisError("DIAGNOSIS_DECISION_BASIS")
|
|
validate_evaluation(payload, context)
|
|
except (DiagnosisError, CaseBuildError) as error:
|
|
code = error.code
|
|
if attempt < policy:
|
|
ref = base + f"/errors/{attempt:04d}.json"
|
|
result = self.store.advance(case_id, package_id, State.BUILDING_DIAGNOSIS,
|
|
step_id=prefix+f":rejected:{attempt}", details={"operation": "diagnosis-rejected", "error_ref": ref, "llm_result_ref": outcome.result_ref, "code": code},
|
|
additions={ref: json_bytes({"code": code, "llm_result_ref": outcome.result_ref})}, expected_generation=reader.manifest["package_generation"])
|
|
self.runner.logger.error("Final diagnosis rejected case_id=%s code=%s", case_id, code)
|
|
return DiagnosisResult("REJECTED", result, code=code)
|
|
payload = self._fallback(context)
|
|
fallback = True
|
|
from .early_stop import accepts_evaluation
|
|
if early and not accepts_evaluation(payload,context):
|
|
return self._decline(reader,context,context_ref,outcome.result_ref,payload,base)
|
|
return self._publish(reader, context, context_ref, outcome.result_ref, payload, operation, attempt, fallback)
|
|
|
|
def _decline(self,reader,context,context_ref,call_ref,payload,base):
|
|
import copy
|
|
state=copy.deepcopy(context['investigation'])
|
|
ref=base+'/review.json'
|
|
review={'status':'declined','context_ref':context_ref,'llm_result_ref':call_ref,'evaluation':payload}
|
|
state['completed']=False;state['countercheck_complete']=False;state['summary_ref']=None
|
|
state['early_stop'].update(status='declined',decline_reason=payload['reason'],review_ref=ref)
|
|
state['hypotheses'][state['early_stop']['hypothesis_id']]['notes'].append(
|
|
{'kind':'assessment','text':('Early diagnosis declined: '+payload['reason'])[:4000],'llm_ref':call_ref})
|
|
_,result=self.loop._save(state,target=State.INVESTIGATING_HYPOTHESES,
|
|
artifact_updates={'diagnosis_context':None},additions={ref:json_bytes(review)})
|
|
self.runner.logger.info('Early diagnosis declined; investigation resumed case_id=%s',state['case_id'])
|
|
return DiagnosisResult('CONTINUED',result.package,code='EARLY_DIAGNOSIS_DECLINED')
|
|
|
|
def _block(self, case_id, package_id, revision, operation, code):
|
|
due = (self.clock() + timedelta(seconds=self.gateway.config.blocked_retry_seconds)).isoformat()
|
|
path = self.store.advance(case_id, package_id, State.BLOCKED, step_id=f"diagnosis:v1:{revision}:config:{key(operation)}",
|
|
reason="LLM_CONFIGURATION", retry_not_before=due, details={"operation": "diagnosis-provider-blocked", "code": code, "automatic_retry": False})
|
|
self.runner.logger.error("Final diagnosis blocked case_id=%s code=%s", case_id, code)
|
|
return DiagnosisResult("BLOCKED", path, code=code, retry_not_before=due)
|
|
|
|
def _request(self,context,reader,attempt,*,modern):
|
|
from .early_stop import namespace
|
|
prefix,_=namespace(reader.manifest['analysis_revision'],context['investigation'])
|
|
failures=[cp for _,cp in reversed(checkpoint_chain(reader))
|
|
if cp['step_id'].startswith(prefix+':rejected:') and int(cp['step_id'].rsplit(':',1)[1])<attempt]
|
|
feedback=[{'code':cp['details']['code'],'llm_result_ref':cp['details']['llm_result_ref'],
|
|
'guidance': 'Correct the cited validation failure against the output schema. Shared evidence may check an alternative; cite actual findings and explain relevance. Do not force confirmation or remove genuine uncertainty.'}
|
|
for cp in failures] if modern else None
|
|
return build_prompt(context,modern=modern,feedback=feedback)
|
|
|
|
def _read_call(self, reader, ref):
|
|
return read_json(reader, ref, "llm-call", self.gateway.config.max_request_bytes + self.gateway.config.max_response_bytes + 16384)
|
|
|
|
def _bind(self, call, reader, context, ref, operation):
|
|
with reader._archive.open(ref.rsplit("/", 1)[0] + "/request.json") as stream:
|
|
raw = reader.validator._read_bounded(stream, min(self.gateway.config.max_request_bytes, self.store.limits.max_file_bytes))
|
|
saved = parse_manifest(raw)
|
|
if not isinstance(saved, dict):
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
attempt=int(operation.split('attempt:',1)[1].split(':',1)[0])
|
|
modern=saved.get('prompt_version') in {'diagnosis_v2','diagnosis_early_v2'}
|
|
expected = self._request(context,reader,attempt,modern=modern)
|
|
expected["privacy"] = saved.get("privacy")
|
|
if (saved != expected or hashlib.sha256(raw).hexdigest() != call["input_request_hash"] or call["operation_key"] != key(f"{reader.manifest['analysis_revision']}:{operation}")
|
|
or call["phase"] != State.BUILDING_DIAGNOSIS or any(call[f] != reader.manifest[f] for f in ("case_id", "package_id", "analysis_revision"))):
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
return modern
|
|
|
|
def _fallback(self, context):
|
|
payload = {"status": "inconclusive", "selected_hypothesis": None, "root_cause": None, "confidence": "unassessed",
|
|
"evaluations": [{"hypothesis_id": hid, "verdict": "unresolved", "reason": "Final model evaluation did not pass validation.", "evidence_refs": [],
|
|
"countercheck": {"adequate": False, "reason": "No accepted final countercheck assessment.", "evidence_refs": []}} for hid in context["investigation"]["hypotheses"]],
|
|
"remaining_uncertainty": ["Final model output rejected repeatedly; no root cause accepted."], "reason": "Final evaluation validation attempts exhausted.",
|
|
"recommended_human_actions": ["Review stored investigation evidence and rejected final evaluations."]}
|
|
|
|
if context.get('evaluation_protocol')==2:
|
|
payload['decision_basis']={'causal_explanation':'No final causal explanation passed validation.',
|
|
'scope_limits':['Evaluation attempts exhausted; no additional research performed.'],
|
|
'sufficient_for_change':False,'contradictions':[]}
|
|
return payload
|
|
|
|
def _publish(self, reader, context, context_ref, call_ref, payload, operation, attempt, fallback=False):
|
|
validate_evaluation(payload, context)
|
|
m = reader.manifest
|
|
base = f"analysis/diagnosis/{m['analysis_revision']:04d}"
|
|
ref, summary_ref = base + "/diagnosis_result.json", base + "/diagnosis_result.md"
|
|
document = {"format": "viano-diagnosis-result", "format_version": 1, **{f: m[f] for f in ("case_id", "package_id", "analysis_revision")},
|
|
"context_ref": context_ref, "context_sha256": digest(context), "llm_result_ref": call_ref, "diagnosis": payload}
|
|
validate(document, "diagnosis-result")
|
|
additions = {ref: json_bytes(document), summary_ref: render_diagnosis(document).encode()}
|
|
failure_ref = None
|
|
if payload["status"] == "inconclusive":
|
|
failure_ref = base + "/failure_report.json"
|
|
failure = {"format": "viano-failure-report", "format_version": 1, **{f: m[f] for f in ("case_id", "package_id", "analysis_revision")},
|
|
"diagnosis_ref": ref, "reason": payload["reason"], "hypothesis_end_states": context["investigation"]["hypotheses"],
|
|
"remaining_questions": {hid: [q for q in h["questions"] if not q["resolved"]] for hid, h in context["investigation"]["hypotheses"].items()},
|
|
"recommended_human_actions": payload["recommended_human_actions"]}
|
|
validate(failure, "failure-report")
|
|
additions[failure_ref] = json_bytes(failure)
|
|
if any(len(data) > self.limits.max_json_bytes for data in additions.values()):
|
|
raise DiagnosisError("DIAGNOSIS_OUTPUT_LIMIT")
|
|
artifacts = {"diagnosis": ref, "diagnosis_summary": summary_ref, "failure_report": failure_ref, "change_request": None,
|
|
"change_request_summary": None, "output_validation": None}
|
|
details = {"operation": "diagnosis-complete", "context_ref": context_ref, "context_sha256": digest(context), "diagnosis_ref": ref,
|
|
"diagnosis_sha256": digest(document), "summary_ref": summary_ref, "failure_ref": failure_ref, "llm_result_ref": call_ref,
|
|
"llm_operation": operation, "attempt": attempt, "fallback": fallback}
|
|
target = State.FAILED_TO_ANALYZE if failure_ref else State.BUILDING_DIAGNOSIS
|
|
result = self.store.advance(m["case_id"], m["package_id"], target, step_id=f"diagnosis:v1:{m['analysis_revision']}:complete",
|
|
reason="DIAGNOSIS_INCONCLUSIVE" if failure_ref else None, details=details, additions=additions, artifacts=artifacts, expected_generation=m["package_generation"])
|
|
return DiagnosisResult("INCONCLUSIVE" if failure_ref else "COMPLETED", result, ref, failure_ref)
|
|
|
|
def _reuse(self, reader, checkpoint, context):
|
|
d = checkpoint["details"]
|
|
document = self._read(reader, d["diagnosis_ref"], "diagnosis-result")
|
|
if digest(document) != d["diagnosis_sha256"] or document["context_sha256"] != digest(context) or d["context_sha256"] != digest(context):
|
|
raise DiagnosisError("DIAGNOSIS_OUTPUT_CHANGED")
|
|
if any(document[f] != reader.manifest[f] for f in ("case_id", "package_id", "analysis_revision")) or document["context_ref"] != d["context_ref"] or document["llm_result_ref"] != d["llm_result_ref"]:
|
|
raise DiagnosisError("DIAGNOSIS_IDENTITY")
|
|
validate_evaluation(document["diagnosis"], context)
|
|
if context['investigation'].get('early_stop',{}).get('status')=='requested':
|
|
from .early_stop import accepts_evaluation
|
|
if not accepts_evaluation(document['diagnosis'],context):
|
|
raise DiagnosisError('DIAGNOSIS_EARLY_ACCEPTANCE')
|
|
call = self._read_call(reader, d["llm_result_ref"])
|
|
modern=self._bind(call, reader, context, d["llm_result_ref"], d["llm_operation"])
|
|
if d["fallback"]:
|
|
if d["attempt"] != context["max_invalid_responses"] or document["diagnosis"] != self._fallback(context):
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
if call["status"] == "SUCCESS":
|
|
try:
|
|
if modern and "decision_basis" not in call["outcome"]["document"]:
|
|
raise DiagnosisError("DIAGNOSIS_DECISION_BASIS")
|
|
validate_evaluation(call["outcome"]["document"], context)
|
|
except (CaseBuildError, DiagnosisError):
|
|
pass
|
|
else:
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
elif call["status"] != "ERROR" or call.get("code") in PROVIDER_CONFIGURATION_ERRORS:
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
elif call["status"] != "SUCCESS" or call["outcome"]["document"] != document["diagnosis"]:
|
|
raise DiagnosisError("DIAGNOSIS_LLM_BINDING")
|
|
with reader._archive.open(d["summary_ref"]) as stream:
|
|
summary = reader.validator._read_bounded(stream, min(self.limits.max_json_bytes, self.store.limits.max_file_bytes))
|
|
if summary != render_diagnosis(document).encode():
|
|
raise DiagnosisError("DIAGNOSIS_OUTPUT_CHANGED")
|
|
for field, ref in (("diagnosis", d["diagnosis_ref"]), ("diagnosis_summary", d["summary_ref"]), ("failure_report", d["failure_ref"])):
|
|
if field == "failure_report" and ref is None:
|
|
# A confirmed cause may still lack verified implementation context.
|
|
# The later CR stage then owns a separate, checkpoint-bound review.
|
|
later = next((cp for _, cp in checkpoint_chain(reader)
|
|
if cp["step_id"] == f"change-request:v1:{document['analysis_revision']}:complete"), None)
|
|
if later and later["details"].get("failure_ref") is not None:
|
|
ref = later["details"]["failure_ref"]
|
|
if reader.manifest["artifacts"].get(field) != ref:
|
|
raise DiagnosisError("DIAGNOSIS_POINTER_CHANGED")
|
|
if d["failure_ref"]:
|
|
failure = self._read(reader, d["failure_ref"], "failure-report")
|
|
if (failure["diagnosis_ref"] != d["diagnosis_ref"] or failure["hypothesis_end_states"] != context["investigation"]["hypotheses"]
|
|
or failure["recommended_human_actions"] != document["diagnosis"]["recommended_human_actions"]
|
|
or failure["reason"] != document["diagnosis"]["reason"]
|
|
or any(failure[f] != document[f] for f in ("case_id", "package_id", "analysis_revision"))
|
|
or failure["remaining_questions"] != {hid: [q for q in h["questions"] if not q["resolved"]] for hid, h in context["investigation"]["hypotheses"].items()}):
|
|
raise DiagnosisError("DIAGNOSIS_OUTPUT_CHANGED")
|
|
return DiagnosisResult("INCONCLUSIVE" if d["failure_ref"] else "COMPLETED", reader.path, d["diagnosis_ref"], d["failure_ref"], True)
|