import copy import hashlib import re from dataclasses import dataclass from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain from viano_bug_analysis.lifecycle.storage import json_bytes from viano_bug_analysis.package.validator import CHUNK_SIZE from .contracts import read_json, require_file, validate from .errors import CaseBuildError from .expected_behavior import ExpectedBehaviorResolver, collect_expected_behavior from .report_collector import collect_statements, provenance from .reproduction_collector import collect_reproductions def _stable_records(previous: list[dict], current: list[dict], prefix: str) -> list[dict]: def identity(record): source = record["provenance"] return source["source_ref"], source["json_pointer"] remaining = {identity(record): record for record in current} result = [] ids = set() for old in previous: if not re.fullmatch(prefix + r"[0-9]{4,}", old["id"]) or old["id"] in ids: raise CaseBuildError("PREVIOUS_IDS", "Prior case contains invalid or duplicate record IDs") ids.add(old["id"]) record = remaining.pop(identity(old), None) if record is None: raise CaseBuildError("PREVIOUS_RECORD_LOST", "Prior source record disappeared from its structured input") record["id"] = old["id"] result.append(record) next_id = max((int(value[len(prefix):]) for value in ids), default=0) + 1 for record in current: if identity(record) in remaining: record["id"] = f"{prefix}{next_id:04d}" next_id += 1 result.append(record) del remaining[identity(record)] return result @dataclass(frozen=True) class CaseBuilderLimits: """Operational limits, independent of the JSON format contracts.""" max_json_bytes: int = 1024**2 max_structured_bytes: int = 8 * 1024**2 max_documents: int = 256 max_case_bytes: int = 8 * 1024**2 max_summary_bytes: int = 8 * 1024**2 max_inline_fact_bytes: int = 16 * 1024 def __post_init__(self): for value in vars(self).values(): if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ValueError("CaseBuilder limits must be positive integers") class CaseAssembler: def __init__(self, limits: CaseBuilderLimits | None = None): self.limits = limits or CaseBuilderLimits() def assemble(self, reader) -> dict: manifest = reader.manifest if manifest["analysis_revision"] < 1: raise CaseBuildError("INVALID_REVISION", "Case construction requires an active analysis revision") used_bytes = 0 documents = 0 def load(path, contract): nonlocal used_bytes, documents require_file(reader, path) if contract == "case": return read_json(reader, path, contract, self.limits.max_case_bytes) documents += 1 used_bytes += reader._archive.getinfo(path).file_size if documents > self.limits.max_documents or used_bytes > self.limits.max_structured_bytes: raise CaseBuildError("STRUCTURED_INPUT_LIMIT", "Declared structured inputs exceed configured limits", path) return read_json(reader, path, contract, self.limits.max_json_bytes) chain = checkpoint_chain(reader) excluded = {"manifest.json"} | {ref for ref, _ in chain} prior_refs = [] for _, checkpoint in reversed(chain): details = checkpoint["details"] if details.get("operation") == "early-correction": ref = details.get("artifact_ref") if not isinstance(ref, str) or not ref.startswith(("analysis/interpretations/", "analysis/hypotheses/")) or "/corrections/" not in ref: raise CaseBuildError("PREVIOUS_CASE", "Invalid correction artifact checkpoint") doc = load(ref, "early-correction") from viano_bug_analysis.analysis.correction import base, digest, PHASES phase = details.get("phase") expected = base(phase, checkpoint["analysis_revision"]) + "/" + ( f"attempt-{doc.get('attempt', 0):04d}.json" if doc["kind"] == "attempt" else doc["kind"] + ".json") if phase in PHASES else None if (ref != expected or digest(doc) != details.get("artifact_sha256") or doc["phase"] != phase or doc["analysis_revision"] != checkpoint["analysis_revision"] or any(doc[f] != manifest[f] for f in ("case_id", "package_id"))): raise CaseBuildError("PREVIOUS_CASE", "Invalid correction artifact checkpoint") excluded.add(ref) if details.get("operation") == "entity-resolution": from viano_bug_analysis.analysis.entity_resolution import digest ref = details.get("artifact_ref") prefix = f"analysis/interpretations/{checkpoint['analysis_revision']:04d}/entity-resolution/" doc = load(ref, "entity-resolution") if (ref != prefix + doc["stage"] + ".json" or digest(doc) != details.get("artifact_sha256") or doc["analysis_revision"] != checkpoint["analysis_revision"] or any(doc[k] != manifest[k] for k in ("case_id", "package_id"))): raise CaseBuildError("PREVIOUS_CASE", "Invalid entity resolution checkpoint") excluded.add(ref) if doc["stage"] == "search": for capture_ref in doc["payload"]["capture_refs"]: capture = read_json(reader, capture_ref, "project-capture", self.limits.max_json_bytes) if capture_ref != prefix + "captures/" + hashlib.sha256(reader.read_bytes(capture_ref)).hexdigest() + ".json": raise CaseBuildError("PREVIOUS_CASE", "Invalid entity resolution capture") excluded.add(capture_ref) if details.get("operation") in {"change-request-start", "change-request-complete", "change-request-rejected", "output-validated"}: fields = ("context_ref", "decision_ref", "change_request_ref", "summary_ref", "failure_ref", "error_ref", "validation_ref", "human_ref") for field in fields: ref = details.get(field) if ref is None: continue if not isinstance(ref, str) or not ref.startswith(("analysis/change-requests/", "handoffs/coding/", "handoffs/human-dev/", "analysis/output/")): raise CaseBuildError("PREVIOUS_CASE", "Invalid output artifact checkpoint") require_file(reader, ref) excluded.add(ref) if details.get("operation") in {"diagnosis-start", "diagnosis-complete", "diagnosis-rejected"}: for field in ("context_ref", "diagnosis_ref", "summary_ref", "failure_ref", "error_ref"): ref = details.get(field) if ref is None: continue if not isinstance(ref, str) or not ref.startswith("analysis/diagnosis/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid diagnosis checkpoint") require_file(reader, ref) excluded.add(ref) if details.get('early_diagnosis_declined'): ref=details.get('early_review_ref') if not isinstance(ref,str) or not ref.startswith('analysis/diagnosis/'): raise CaseBuildError('PREVIOUS_CASE','Invalid early review checkpoint') require_file(reader,ref) excluded.add(ref) if details.get("operation")=="investigation-state": for field in ("state_ref","context_ref","summary_ref","working_brief_ref"): ref=details.get(field) if ref is None:continue if not isinstance(ref,str) or not ref.startswith("analysis/investigation/"): raise CaseBuildError("PREVIOUS_CASE","Invalid investigation checkpoint") require_file(reader,ref) if field == 'working_brief_ref': brief = read_json(reader, ref, 'investigation-working-brief-record', 4 * 1024 * 1024) if any(brief[k] != checkpoint[k] for k in ('case_id','package_id','analysis_revision')): raise CaseBuildError('PREVIOUS_CASE', 'Invalid working brief identity') excluded.add(ref) if details.get("operation") == "evidence-stored": if details.get("result_ref"): tool_result = read_json(reader, details["result_ref"], "investigation-result", 4 * 1024 * 1024) from viano_bug_analysis.tools.project_query import ProjectResearch for capture_ref, data in ProjectResearch.additions(tool_result).items(): require_file(reader, capture_ref) if reader.read_bytes(capture_ref) != data: raise CaseBuildError("PREVIOUS_CASE", "Project capture changed") excluded.add(capture_ref) refs = [details.get("index_ref"), details.get("request_ref"), details.get("result_ref"), *details.get("record_refs", [])] for ref in refs: if ref is None: continue if not isinstance(ref, str) or not ref.startswith("analysis/evidence/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid evidence checkpoint") require_file(reader, ref) excluded.add(ref) for ref in details.get("record_refs", []): record = read_json(reader, ref, "evidence-record", self.limits.max_json_bytes) content_ref = record["content_ref"] if not content_ref.startswith("analysis/evidence/content/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid evidence content reference") require_file(reader, content_ref) excluded.add(content_ref) if details.get("operation") in {"async-test-wait", "async-test-feedback-confirmed"}: for field in ("request_ref", "feedback_ref"): ref = details.get(field) if ref is None: continue if not isinstance(ref, str) or not ref.startswith("analysis/async-tests/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid async test checkpoint") require_file(reader, ref) excluded.add(ref) if field == "feedback_ref": feedback = load(ref, "async-test-feedback") for attachment in feedback["attachments"]: path = attachment["source_ref"] require_file(reader, path) if path.startswith("analysis/async-tests/"): excluded.add(path) if details.get("operation") == "budget-event": ref = details.get("event_ref") if not isinstance(ref, str) or not ref.startswith("analysis/budgets/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid budget artifact checkpoint") require_file(reader, ref) excluded.add(ref) if details.get("operation") in {"hypotheses-generating", "hypotheses-built", "hypotheses-rejected"}: for field in ("context_ref", "policy_ref", "hypotheses_ref", "registry_ref", "summary_ref", "error_ref"): if field not in details: continue ref = details[field] if not isinstance(ref, str) or not ref.startswith("analysis/hypotheses/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid hypothesis artifact checkpoint") require_file(reader, ref) excluded.add(ref) if details.get("operation") in {"case-interpreting", "case-interpreted", "case-interpretation-rejected"}: for field in ("context_ref", "interpretation_ref", "summary_ref", "error_ref"): if field not in details: continue ref = details[field] if not isinstance(ref, str) or not ref.startswith("analysis/interpretations/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid interpretation artifact checkpoint") require_file(reader, ref) excluded.add(ref) if details.get("operation") == "llm-round": for field in ("request_ref", "result_ref"): ref = details.get(field) if not isinstance(ref, str) or not ref.startswith("llm/operations/"): raise CaseBuildError("PREVIOUS_CASE", "Invalid LLM artifact checkpoint") require_file(reader, ref) excluded.add(ref) if details.get("operation") == "case-built" and details.get("builder_version") == "1": for field in ("case_ref", "summary_ref"): if not isinstance(details.get(field), str): raise CaseBuildError("PREVIOUS_CASE", "Incomplete case completion checkpoint") require_file(reader, details[field]) excluded.add(details[field]) if details.get("case_ref") not in prior_refs: prior_refs.append(details["case_ref"]) prior_ref = manifest["artifacts"].get("case") if prior_ref: excluded.add(prior_ref) if prior_ref not in prior_refs: prior_refs.append(prior_ref) summary_ref = manifest["artifacts"].get("case_summary") if summary_ref: excluded.add(summary_ref) previous = load(prior_ref, "case") if prior_ref else None if previous and (previous["case"]["case_id"] != manifest["case_id"] or previous["case"]["package_id"] != manifest["package_id"] or previous["case"]["analysis_revision"] > manifest["analysis_revision"]): raise CaseBuildError("CASE_IDENTITY", "Previous case has inconsistent identity or revision", prior_ref) inventory = {} for name in sorted(reader.files): if name == "manifest.json" or name in {ref for ref, _ in chain}: continue digest = hashlib.sha256() size = 0 with reader._archive.open(name) as stream: while chunk := stream.read(CHUNK_SIZE): size += len(chunk) if size > reader.validator.limits.max_file_bytes: raise CaseBuildError("INPUT_LIMIT", "Payload exceeds configured file limit", name) digest.update(chunk) inventory[name] = {"path": name, "sha256": digest.hexdigest(), "size": size} index_ref = manifest["artifacts"].get("case_inputs") index = load(index_ref, "case-inputs") if index_ref else {} roles = [] def add(path, role): require_file(reader, path) if path in excluded: raise CaseBuildError("INPUT_REFERENCE", "Generated case or checkpoint cannot serve as raw input", path) if (path, role) not in roles: roles.append((path, role)) if previous: for record in previous["inputs"]["records"]: add(record["path"], record["role"]) actual = inventory[record["path"]] if record["sha256"] != actual["sha256"] or record["size"] != actual["size"]: raise CaseBuildError("PREVIOUS_INPUT_CHANGED", "A source referenced by the prior case changed", record["path"]) if index_ref: add(index_ref, "index") for field, role in (("reports", "report"), ("reproductions", "reproduction"), ("attachments", "attachment"), ("expected_research", "expected_research")): paths = index.get(field, []) if len(paths) != len(set(paths)): raise CaseBuildError("INPUT_DUPLICATE", "A source is listed twice in the same role", index_ref) for path in paths: add(path, role) for role, paths in index.get("context", {}).items(): if role not in {"runtime", "database", "source", "research"}: continue if len(paths) != len(set(paths)): raise CaseBuildError("INPUT_DUPLICATE", "A context source is listed twice", index_ref) for path in paths: add(path, role) reports = [(path, load(path, "report")) for path, role in roles if role == "report"] reproductions = [(path, load(path, "reproduction")) for path, role in roles if role == "reproduction"] for _, document in reports + reproductions: nodes = [document] + [node for field in ("observations", "expected_behavior", "comments", "tester_hypotheses", "facts", "entities", "data_quality_issues") for node in document.get(field, [])] for node in nodes: if "origin" in node: for field in ("source_ref", "archive_ref"): if node["origin"].get(field) is not None: require_file(reader, node["origin"][field]) for _, document in reproductions: for path in document.get("attachments", []): add(path, "attachment") if document.get("events_ref"): add(document["events_ref"], "attachment") observations = collect_statements(reports + reproductions, "observations", "O") comments = collect_statements(reports, "comments", "C") hypotheses = collect_statements(reports, "tester_hypotheses", "T") expected = collect_expected_behavior(reports) reproduction_records, facts = collect_reproductions(reproductions, self.limits.max_inline_fact_bytes) research_documents = [(path, load(path, "research-result")) for path, role in roles if role == "expected_research"] expected = ExpectedBehaviorResolver().resolve(expected, research_documents) if previous: old_humans = previous["human_observations"] observations = _stable_records(old_humans["observations"], observations, "O") comments = _stable_records(old_humans["comments"], comments, "C") hypotheses = _stable_records(old_humans["tester_hypotheses"], hypotheses, "T") expected["statements"] = _stable_records(previous["expected_behavior"]["statements"], expected["statements"], "E") facts = _stable_records(previous["runtime_facts"]["facts"], facts, "F") entities = {} for path, document in reports + reproductions: for index, entity in enumerate(document.get("entities", [])): entity_id = entity["id"] if isinstance(entity["id"], str) else str(int(entity["id"])) identity = (entity["type"], entity_id) record = entities.setdefault(identity, {"type": identity[0], "id": identity[1], "roles": [], "provenance": []}) record["roles"] = sorted(set(record["roles"]) | set(entity.get("roles", []))) record["provenance"].append(provenance(path, f"/entities/{index}", "report" if document["format"] == "viano-bug-report" else "reproduction_record", entity)) research_request = ExpectedBehaviorResolver().request(manifest, [d for _, d in reports], [entities[k] for k in sorted(entities)], collect_expected_behavior(reports)) for path, result in research_documents: request_ref = path.rsplit("/", 1)[0] + "/request.json" request = load(request_ref, "research-request") add(request_ref, "research") if hashlib.sha256(json_bytes(request)).hexdigest() != result["request_hash"]: raise CaseBuildError("RESEARCH_REQUEST_MISMATCH", "Stored research request hash does not match", path) if request["case_id"] != manifest["case_id"] or request["package_id"] != manifest["package_id"] or request["analysis_revision"] > manifest["analysis_revision"]: raise CaseBuildError("RESEARCH_IDENTITY", "Research belongs to another case or future revision", path) issues = [] def issue(code, message, refs=None): issues.append({"code": code, "message": message, "refs": refs or []}) for path, report in reports: for index, item in enumerate(report.get("data_quality_issues", [])): source = item.get("origin", {"source_ref": path, "json_pointer": f"/data_quality_issues/{index}"}) issue(item["code"], item["message"], [{"source_ref": source["source_ref"], "json_pointer": source["json_pointer"]}]) if not index_ref: issue("MISSING_INPUT_INDEX", "No explicit case input index was supplied") if not reports: issue("MISSING_REPORT", "No structured report was supplied") if not observations: issue("MISSING_OBSERVATIONS", "No direct observations were supplied") if not reproductions: issue("MISSING_REPRODUCTION", "No structured reproduction was supplied") if expected["completeness"] not in {"reported_complete", "researched_complete"}: issue("EXPECTED_BEHAVIOR_" + expected["completeness"].upper(), "Expected-behavior completeness is " + expected["completeness"], [{"source_ref": path, "json_pointer": "/expected_behavior_complete"} for path, report in reports if "expected_behavior_complete" in report]) for category, records in (("observation", observations), ("expected_behavior", expected["statements"])): claims = {} for record in records: if "claim" in record: claim = record["claim"] claims.setdefault(claim["key"], []).append(record) for claim_key, records in sorted(claims.items()): if len({json_bytes(record["claim"]["value"]) for record in records}) > 1: refs = [{"source_ref": r["provenance"]["source_ref"], "json_pointer": r["provenance"]["json_pointer"]} for r in records] issue("CONFLICTING_CLAIMS", f"Different explicit values for {category} claim {claim_key}", refs) if category == "expected_behavior": expected["completeness"] = "conflicting" if expected["completeness"] == "researched_complete": research_request = None if expected["completeness"] == "conflicting": tester_expected = collect_expected_behavior(reports) tester_expected["completeness"] = "conflicting" research_request = ExpectedBehaviorResolver().request(manifest, [d for _, d in reports], [entities[k] for k in sorted(entities)], tester_expected) records = [inventory[path] | {"role": role} for path, role in roles] classified = {path for path, _ in roles} raw_inventory = [value for name, value in inventory.items() if name not in excluded] request = {"builder_version": "1", "case_id": manifest["case_id"], "package_id": manifest["package_id"], "analysis_revision": int(manifest["analysis_revision"]), "project": manifest["project"], "index_ref": index_ref, "payloads": raw_inventory, "roles": roles} request_hash = hashlib.sha256(json_bytes(request)).hexdigest() refs_for = lambda role: [inventory[path] for path, value in roles if value == role] case = { "format": "viano-case", "format_version": 1, "builder_version": "1", "case": {"case_id": manifest["case_id"], "package_id": manifest["package_id"], "analysis_revision": int(manifest["analysis_revision"]), "source_generation": int(manifest["package_generation"]), "project": copy.deepcopy(manifest["project"])}, "bug": {"reports": [{"source_ref": path, "title": document.get("title"), "summary": document.get("summary"), **({"origin": copy.deepcopy(document["origin"])} if "origin" in document else {})} for path, document in reports]}, "human_observations": {"observations": observations, "comments": comments, "tester_hypotheses": hypotheses}, "expected_behavior": expected, "reproduction": {"records": reproduction_records}, "entities": [entities[identity] for identity in sorted(entities)], "runtime_facts": {"facts": facts, "raw_refs": refs_for("runtime")}, "database_context": {"refs": refs_for("database")}, "source_context": {"refs": refs_for("source")}, "research": {"refs": refs_for("research") + refs_for("expected_research"), "performed": bool(research_documents), "request": research_request}, "data_quality": {"issues": issues, "coverage": {"reports": len(reports), "reproductions": len(reproductions), "observations": len(observations)}}, "inputs": {"index_ref": index_ref, "records": records, "unclassified": [value for name, value in inventory.items() if name not in classified and name not in excluded], "prior_cases": [inventory[ref] for ref in prior_refs], "request_hash": request_hash}, } validate(case, "case") if len(json_bytes(case)) > self.limits.max_case_bytes: raise CaseBuildError("CASE_LIMIT", "Assembled case exceeds configured byte limit") return case