Add change request builder and validated local output publication

src/viano_bug_analysis/analysis/change_request_contracts.py

@@ -0,0 +1,88 @@
+"""Deterministic coding policy and reference gates; never model-editable."""
+import html
+import json
+from importlib.resources import files
+
+from viano_bug_analysis.case.contracts import validate
+from viano_bug_analysis.lifecycle.storage import json_bytes
+
+
+class ChangeRequestError(Exception):
+    def __init__(self, code):
+        self.code = code
+        super().__init__(code)
+
+
+def bugfix_policy():
+    return {"version": 1, "type": "bugfix", "precedence": "mandatory_over_case_guidance",
+            "rules": ["Prefer the smallest practically sufficient change.", "Do not add features.",
+                      "Do not perform unnecessary refactoring.", "Preserve behavior outside the diagnosed defect.",
+                      "Do not change unrelated files or database rows.", "Do not invent APIs, schemas or symbols.",
+                      "Use only locally verified project context.", "Report missing implementation context instead of guessing.",
+                      "Do not deploy, restart services or access production."]}
+
+
+def build_prompt(context):
+    schema = json.loads(files("viano_bug_analysis.schemas").joinpath("change-request-output-v1.schema.json").read_text())
+    return {"prompt_version": "change_request_v1", "privacy": "configured", "output_schema": schema,
+            "messages": [{"role": "system", "content": files("viano_bug_analysis.llm.prompts").joinpath("change_request_v1.txt").read_text(encoding="utf-8")},
+                         {"role": "user", "content": json_bytes({"task": "build_change_request", "input": context, "output_schema": schema}).decode()}]}
+
+
+def item_source(item):
+    source = item.get("source")
+    return item.get("source_ref") or (source.get("source_ref") if isinstance(source, dict) else None)
+
+
+def validate_request(payload, context):
+    validate(payload, "change-request-output")
+    evidence = context["evidence"]
+    local = context["local_context"]
+    def refs(refs):
+        if any(eid not in evidence for eid in refs):
+            raise ChangeRequestError("CHANGE_REQUEST_EVIDENCE")
+    paths = set()
+    for scope in payload["affected_scope"]:
+        ref = scope["source_ref"]
+        if ref in paths or ref not in local or local[ref]["kind"] != scope["kind"]:
+            raise ChangeRequestError("CHANGE_REQUEST_SCOPE")
+        paths.add(ref)
+        refs(scope["evidence_refs"])
+        if not any(item_source(item) == ref for eid in scope["evidence_refs"] for item in evidence[eid]["content"]["items"]):
+            raise ChangeRequestError("CHANGE_REQUEST_SCOPE_EVIDENCE")
+    for check in payload["validation_requirements"]:
+        refs(check["evidence_refs"])
+    if payload["status"] == "insufficient_context":
+        if not payload["missing_information"]:
+            raise ChangeRequestError("CHANGE_REQUEST_MISSING_INFORMATION")
+        return
+    if not paths or payload["missing_information"] or not payload["validation_requirements"]:
+        raise ChangeRequestError("CHANGE_REQUEST_INCOMPLETE")
+    root_refs = context["diagnosis"]["diagnosis"]["root_cause"]["evidence_refs"]
+    if not any(item_source(item) in paths for eid in root_refs for item in evidence[eid]["content"]["items"]):
+        raise ChangeRequestError("CHANGE_REQUEST_ROOT_SCOPE")
+
+
+def make_document(context, payload, context_ref, call_ref):
+    from viano_bug_analysis.evidence.store import digest
+    return {"format": "viano-change-request", "format_version": 1,
+            **{f: context[f] for f in ("case_id", "package_id", "analysis_revision")},
+            "task": {"type": "bugfix", "id": f"CR-{context['analysis_revision']:04d}"},
+            "diagnosis_ref": context["diagnosis_ref"], "diagnosis_sha256": context["diagnosis_sha256"],
+            "context_ref": context_ref, "context_sha256": digest(context), "llm_result_ref": call_ref,
+            "policy": bugfix_policy(), "current_behavior": context["interpretation"]["observed_behavior"],
+            "expected_behavior": context["interpretation"]["expected_behavior"],
+            "root_cause": context["diagnosis"]["diagnosis"]["root_cause"],
+            "project_context_refs": [local for ref, local in sorted(context["local_context"].items())
+                                     if ref in {s["source_ref"] for s in payload["affected_scope"]}], "request": payload}
+
+
+def render_change_request(document):
+    lines = ["# Bugfix change request", "", "Task: " + document["task"]["id"], "",
+             "Diagnosis: " + document["diagnosis_ref"], "", "## Mandatory policy", ""]
+    lines += ["- " + html.escape(rule) for rule in document["policy"]["rules"]]
+    fields = {field: document[field] for field in ("current_behavior", "expected_behavior", "root_cause", "project_context_refs")}
+    fields.update(document["request"])
+    for field, value in fields.items():
+        lines += ["", "## " + field.replace("_", " "), "", "```json", json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True).replace("`", "\\u0060"), "```"]
+    return "\n".join(lines) + "\n"