Govern Process Analysis changes in diff and impact

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Process Contract Schema v0.2 with typed transitions, I/O contracts, SLA/KPI, risk, evidence and agent policy metadata.
 - Process Analysis Canvas with observed timing, ownership, variants, evidence-backed pain points, data flows and a standalone HTML renderer.
+- Governed process-analysis references plus semantic diff and impact propagation for variants, pain points and data flows.
 - v0.1 migration command.
 - BPMN supported-subset import and richer export with lanes and parallel gateways.
 - Process governance policy gates.

README.md

@@ -169,7 +169,7 @@ The source-tree schema is [`schemas/process.schema.json`](schemas/process.schema
 
 Process discovery needs more than a flowchart. A useful assessment view must show the real execution details that disappear from high-level BPMN: who performs each step, which system is used, what goes in and out, how much active and waiting time is observed, what variants exist, where evidence-backed pain points occur, how data moves, and who can change the step.
 
-The optional analysis fields keep those facts in the same governed process contract:
+The optional analysis fields keep those facts in the same governed process contract. Analysis references are validated, and changes to variants, pain points, and data flows are included in semantic diff and impact reports:
 
 ```yaml
 steps:

docs/guides/process-change-impact-analysis.md

@@ -50,7 +50,7 @@ process-code impact old.process.yaml new.process.yaml \
   --json
 ```
 
-The impact report starts with changed steps and then derives affected roles, systems, business objects, interfaces, controls, risks, evidence and linked artifacts. Generated test scenarios connected to those changes are included as recommended regression scope.
+The impact report starts with changed steps and then derives affected roles, systems, business objects, interfaces, controls, risks, evidence and linked artifacts. It also tracks changes to process-analysis variants, pain points and data flows by stable ID. When those analysis facts point to steps, systems or evidence, that context is added to the impact report and related step tests can enter the recommended regression scope.
 
 ## Pull-request use
 

docs/specification.md

@@ -79,6 +79,16 @@ The optional top-level `analysis` object records facts that belong to the proces
 - `analysis.pain_points`: evidence-backed observations, causes, impacts, and affected steps.
 - `analysis.data_flows`: system-to-system data movement and field-level source-of-truth ownership.
 
+Analysis items are governed contract entities, not free-form notes:
+
+- IDs must be unique inside each analysis section.
+- `analysis.variants[].path` entries must reference existing step IDs.
+- `analysis.pain_points[].step` must reference an existing step when present.
+- `analysis.pain_points[].evidence` entries must reference the evidence catalog.
+- `analysis.data_flows[].from` and `.to` must reference declared systems.
+
+Semantic diff compares variants, pain points, and data flows by stable ID. Impact analysis also derives the related steps, systems, evidence, and regression scope from changed analysis items. This keeps discovery facts inside the same review loop as process design changes.
+
 A recommended terminology is:
 
 - **Process Discovery**: the activity of observing the real process.

src/process_as_code/diff.py

@@ -4,6 +4,7 @@
 
 
 SECTIONS = ("steps", "roles", "systems", "objects", "interfaces", "controls", "risks", "evidence", "artifacts")
+ANALYSIS_SECTIONS = ("variants", "pain_points", "data_flows")
 
 
 def _by_id(items: list[Any] | None) -> dict[str, dict[str, Any]]:
@@ -14,26 +15,36 @@ def _by_id(items: list[Any] | None) -> dict[str, dict[str, Any]]:
     }
 
 
+def _diff_items(old_items: list[Any] | None, new_items: list[Any] | None) -> dict[str, Any]:
+    before, after = _by_id(old_items), _by_id(new_items)
+    added = sorted(set(after) - set(before))
+    removed = sorted(set(before) - set(after))
+    changed: dict[str, Any] = {}
+    for item_id in sorted(set(before) & set(after)):
+        if before[item_id] == after[item_id]:
+            continue
+        fields: dict[str, Any] = {}
+        for field in sorted(set(before[item_id]) | set(after[item_id])):
+            if before[item_id].get(field) != after[item_id].get(field):
+                fields[field] = {"old": before[item_id].get(field), "new": after[item_id].get(field)}
+        changed[item_id] = fields
+    return {"added": added, "removed": removed, "changed": changed}
+
+
 def semantic_diff(old: dict[str, Any], new: dict[str, Any]) -> dict[str, Any]:
-    result: dict[str, Any] = {"process": {}, "sections": {}}
+    result: dict[str, Any] = {"process": {}, "sections": {}, "analysis": {}}
     old_meta, new_meta = old.get("process", {}), new.get("process", {})
     for field in sorted(set(old_meta) | set(new_meta)):
         if old_meta.get(field) != new_meta.get(field):
             result["process"][field] = {"old": old_meta.get(field), "new": new_meta.get(field)}
 
     for section in SECTIONS:
-        before, after = _by_id(old.get(section)), _by_id(new.get(section))
-        added = sorted(set(after) - set(before))
-        removed = sorted(set(before) - set(after))
-        changed: dict[str, Any] = {}
-        for item_id in sorted(set(before) & set(after)):
-            if before[item_id] != after[item_id]:
-                fields: dict[str, Any] = {}
-                for field in sorted(set(before[item_id]) | set(after[item_id])):
-                    if before[item_id].get(field) != after[item_id].get(field):
-                        fields[field] = {"old": before[item_id].get(field), "new": after[item_id].get(field)}
-                changed[item_id] = fields
-        result["sections"][section] = {"added": added, "removed": removed, "changed": changed}
+        result["sections"][section] = _diff_items(old.get(section), new.get(section))
+
+    old_analysis = old.get("analysis", {}) if isinstance(old.get("analysis"), dict) else {}
+    new_analysis = new.get("analysis", {}) if isinstance(new.get("analysis"), dict) else {}
+    for section in ANALYSIS_SECTIONS:
+        result["analysis"][section] = _diff_items(old_analysis.get(section), new_analysis.get(section))
     return result
 
 
@@ -58,6 +69,24 @@ def diff_markdown(diff: dict[str, Any]) -> str:
         for item_id, fields in changes["changed"].items():
             lines.append(f"- Changed `{item_id}`: {', '.join(f'`{field}`' for field in fields)}")
         lines.append("")
+
+    analysis_started = False
+    for section, changes in diff.get("analysis", {}).items():
+        if not (changes["added"] or changes["removed"] or changes["changed"]):
+            continue
+        changed_any = True
+        if not analysis_started:
+            lines += ["## Process analysis", ""]
+            analysis_started = True
+        lines += [f"### {section.replace('_', ' ').title()}", ""]
+        for item_id in changes["added"]:
+            lines.append(f"- Added `{item_id}`")
+        for item_id in changes["removed"]:
+            lines.append(f"- Removed `{item_id}`")
+        for item_id, fields in changes["changed"].items():
+            lines.append(f"- Changed `{item_id}`: {', '.join(f'`{field}`' for field in fields)}")
+        lines.append("")
+
     if not changed_any:
         lines.append("No semantic changes.")
     return "\n".join(lines).rstrip() + "\n"

src/process_as_code/impact.py

@@ -37,6 +37,37 @@ def _refs_from_step(step: dict[str, Any]) -> dict[str, set[str]]:
     }
 
 
+ANALYSIS_SECTIONS = ("variants", "pain_points", "data_flows")
+
+
+def _analysis_by_id(data: dict[str, Any], section: str) -> dict[str, dict[str, Any]]:
+    analysis = data.get("analysis", {})
+    if not isinstance(analysis, dict):
+        return {}
+    return _by_id(analysis.get(section))
+
+
+def _analysis_refs(section: str, item: dict[str, Any]) -> dict[str, set[str]]:
+    steps: set[str] = set()
+    systems: set[str] = set()
+    evidence: set[str] = set()
+    if section == "variants":
+        path = item.get("path", [])
+        if isinstance(path, list):
+            steps.update(value for value in path if isinstance(value, str))
+    elif section == "pain_points":
+        if isinstance(item.get("step"), str):
+            steps.add(item["step"])
+        refs = item.get("evidence", [])
+        if isinstance(refs, list):
+            evidence.update(value for value in refs if isinstance(value, str))
+    elif section == "data_flows":
+        for field in ("from", "to"):
+            if isinstance(item.get(field), str):
+                systems.add(item[field])
+    return {"steps": steps, "systems": systems, "evidence": evidence}
+
+
 def impact_analysis(old: dict[str, Any], new: dict[str, Any], *, base_dir: str | Path | None = None, resolve_external: bool = False, allow_network: bool = False) -> dict[str, Any]:
     diff = semantic_diff(old, new)
     step_changes = diff["sections"]["steps"]
@@ -49,6 +80,23 @@ def impact_analysis(old: dict[str, Any], new: dict[str, Any], *, base_dir: str |
                 continue
             for name, values in _refs_from_step(source).items():
                 affected[name].update(values)
+
+    analysis_changes: dict[str, list[str]] = {}
+    analysis_affected_steps: set[str] = set()
+    for section in ANALYSIS_SECTIONS:
+        changes = diff.get("analysis", {}).get(section, {"added": [], "removed": [], "changed": {}})
+        changed_ids = set(changes["added"]) | set(changes["removed"]) | set(changes["changed"])
+        analysis_changes[section] = sorted(changed_ids)
+        old_items, new_items = _analysis_by_id(old, section), _analysis_by_id(new, section)
+        for item_id in changed_ids:
+            for source in (old_items.get(item_id), new_items.get(item_id)):
+                if not source:
+                    continue
+                refs = _analysis_refs(section, source)
+                analysis_affected_steps.update(refs["steps"])
+                affected["systems"].update(refs["systems"])
+                affected["evidence"].update(refs["evidence"])
+
     for section in affected:
         if section not in diff["sections"]:
             continue
@@ -57,7 +105,8 @@ def impact_analysis(old: dict[str, Any], new: dict[str, Any], *, base_dir: str |
         affected[section].update(changes["removed"])
         affected[section].update(changes["changed"].keys())
 
-    tests = [test for test in generate_test_scope(new) if test["step"] in changed_steps or any(ref in test["id"] for ref in affected["interfaces"] | affected["controls"] | affected["risks"])]
+    test_steps = changed_steps | analysis_affected_steps
+    tests = [test for test in generate_test_scope(new) if test["step"] in test_steps or any(ref in test["id"] for ref in affected["interfaces"] | affected["controls"] | affected["risks"])]
     risk_flags: list[str] = []
     if affected["controls"]: risk_flags.append("control-change")
     if affected["interfaces"]: risk_flags.append("integration-change")
@@ -74,17 +123,31 @@ def impact_analysis(old: dict[str, Any], new: dict[str, Any], *, base_dir: str |
 
     return {
         "changed_steps": sorted(changed_steps),
+        "analysis_changes": analysis_changes,
+        "analysis_affected_steps": sorted(analysis_affected_steps),
         "affected": {name: sorted(values) for name, values in affected.items()},
         "risk_flags": risk_flags,
         "recommended_tests": tests,
         "resolved_artifacts": resolved,
         "semantic_diff": diff,
     }
 
-
 def impact_markdown(result: dict[str, Any]) -> str:
     lines = ["# Process change impact", "", "## Changed steps", ""]
     lines += [f"- `{step}`" for step in result["changed_steps"]] or ["No step-level changes."]
+
+    analysis_changes = result.get("analysis_changes", {})
+    if any(analysis_changes.values()):
+        lines += ["", "## Process analysis changes", ""]
+        for section, values in analysis_changes.items():
+            if values:
+                rendered = ", ".join(f"`{value}`" for value in values)
+                lines.append(f"- **{section.replace('_', ' ').title()}**: {rendered}")
+        affected_steps = result.get("analysis_affected_steps", [])
+        if affected_steps:
+            rendered = ", ".join(f"`{value}`" for value in affected_steps)
+            lines.append(f"- **Affected steps from analysis**: {rendered}")
+
     lines += ["", "## Affected context", ""]
     for section, values in result["affected"].items():
         rendered = ", ".join(f"`{value}`" for value in values) if values else "—"
@@ -101,5 +164,5 @@ def impact_markdown(result: dict[str, Any]) -> str:
         for test in result["recommended_tests"]:
             lines.append(f"| `{test['id']}` | {test['type']} | {test['scenario']} |")
     else:
-        lines.append("No generated tests are directly linked to the changed steps.")
+        lines.append("No generated tests are directly linked to the changed steps or analysis context.")
     return "\n".join(lines).rstrip() + "\n"

src/process_as_code/validate.py

@@ -57,6 +57,77 @@ def _validate_contracts(result: ValidationResult, sid: str, field_name: str, val
             result.errors.append(f"step '{sid}' {field_name}[{index}] requires id, name or ref")
 
 
+def _validate_analysis(
+    result: ValidationResult,
+    data: dict[str, Any],
+    step_ids: set[str],
+    known: dict[str, set[str]],
+) -> None:
+    analysis = data.get("analysis")
+    if analysis is None:
+        return
+    if not isinstance(analysis, dict):
+        result.errors.append("top-level 'analysis' must be an object")
+        return
+
+    labels = {"variants": "variant", "pain_points": "pain point", "data_flows": "data flow"}
+    for section, label in labels.items():
+        items = analysis.get(section)
+        if items is None:
+            continue
+        if not isinstance(items, list):
+            result.errors.append(f"analysis.{section} must be a list")
+            continue
+        ids: list[str] = []
+        for index, item in enumerate(items):
+            if not isinstance(item, dict):
+                result.errors.append(f"analysis.{section}[{index}] must be an object")
+                continue
+            item_id = item.get("id")
+            if not isinstance(item_id, str) or not item_id.strip():
+                result.errors.append(f"analysis.{section}[{index}].id is required")
+            else:
+                ids.append(item_id)
+
+            if section == "variants":
+                path = item.get("path")
+                if path is not None and not isinstance(path, list):
+                    result.errors.append(f"analysis variant '{item_id or index}' path must be a list")
+                elif isinstance(path, list):
+                    for ref in path:
+                        if not isinstance(ref, str):
+                            result.errors.append(f"analysis variant '{item_id or index}' path must contain step IDs")
+                        elif ref not in step_ids:
+                            result.errors.append(f"analysis variant '{item_id or index}' references unknown step '{ref}'")
+
+            elif section == "pain_points":
+                step = item.get("step")
+                if step is not None and not isinstance(step, str):
+                    result.errors.append(f"analysis pain point '{item_id or index}' step must be a string")
+                elif isinstance(step, str) and step not in step_ids:
+                    result.errors.append(f"analysis pain point '{item_id or index}' references unknown step '{step}'")
+                evidence = item.get("evidence")
+                if evidence is not None and not isinstance(evidence, list):
+                    result.errors.append(f"analysis pain point '{item_id or index}' evidence must be a list")
+                elif isinstance(evidence, list):
+                    for ref in evidence:
+                        if not isinstance(ref, str):
+                            result.errors.append(f"analysis pain point '{item_id or index}' evidence must contain evidence IDs")
+                        elif ref not in known["evidence"]:
+                            result.errors.append(f"analysis pain point '{item_id or index}' references unknown evidence '{ref}'")
+
+            elif section == "data_flows":
+                for field_name in ("from", "to"):
+                    ref = item.get(field_name)
+                    if not isinstance(ref, str) or not ref.strip():
+                        result.errors.append(f"analysis data flow '{item_id or index}' {field_name} is required")
+                    elif ref not in known["systems"]:
+                        result.errors.append(f"analysis data flow '{item_id or index}' references unknown system '{ref}'")
+
+        for duplicate in _duplicates(ids):
+            result.errors.append(f"duplicate analysis {label} id '{duplicate}'")
+
+
 def validate_process(data: dict[str, Any]) -> ValidationResult:
     result = ValidationResult()
     version = data.get("version")
@@ -123,6 +194,8 @@ def validate_process(data: dict[str, Any]) -> ValidationResult:
     if meta.get("owner") and known["roles"] and meta.get("owner") not in known["roles"]:
         result.errors.append(f"process.owner references unknown role '{meta.get('owner')}'")
 
+    _validate_analysis(result, data, step_id_set, known)
+
     for step in steps:
         if not isinstance(step, dict) or not isinstance(step.get("id"), str):
             continue

tests/test_analysis_canvas.py

@@ -1,6 +1,9 @@
+from copy import deepcopy
 from pathlib import Path
 
 from process_as_code.cli import main
+from process_as_code.diff import diff_markdown, semantic_diff
+from process_as_code.impact import impact_analysis, impact_markdown
 from process_as_code.io import load_process
 from process_as_code.render import to_analysis_canvas
 from process_as_code.validate import validate_process
@@ -30,3 +33,47 @@ def test_cli_canvas_writes_standalone_html(tmp_path):
     assert rendered.startswith("<!doctype html>")
     assert "Actual execution" in rendered
     assert "Variants & exceptions" in rendered
+
+
+def test_analysis_references_are_governed():
+    data = load_process(EXAMPLE)
+    broken = deepcopy(data)
+    broken["analysis"]["variants"][0]["path"].append("missing_step")
+    broken["analysis"]["pain_points"][0]["step"] = "missing_step"
+    broken["analysis"]["pain_points"][0]["evidence"].append("missing_evidence")
+    broken["analysis"]["data_flows"][0]["to"] = "missing_system"
+
+    result = validate_process(broken)
+    assert not result.ok
+    errors = "\n".join(result.errors)
+    assert "analysis variant 'high_value_order' references unknown step 'missing_step'" in errors
+    assert "analysis pain point 'slow_customer_search' references unknown step 'missing_step'" in errors
+    assert "analysis pain point 'slow_customer_search' references unknown evidence 'missing_evidence'" in errors
+    assert "analysis data flow 'customer_po_to_s4' references unknown system 'missing_system'" in errors
+
+
+def test_analysis_changes_flow_into_diff_impact_and_test_scope():
+    old = load_process(EXAMPLE)
+    new = deepcopy(old)
+    new["analysis"]["pain_points"][0]["cause"] = "Customer search filters are incomplete"
+    new["analysis"]["pain_points"][0]["evidence"] = ["observed_session_01"]
+    new["analysis"]["variants"][0]["path"].append("check_customer")
+    new["analysis"]["data_flows"][0]["data"].append("sales_org")
+
+    diff = semantic_diff(old, new)
+    assert "slow_customer_search" in diff["analysis"]["pain_points"]["changed"]
+    assert "high_value_order" in diff["analysis"]["variants"]["changed"]
+    assert "customer_po_to_s4" in diff["analysis"]["data_flows"]["changed"]
+    rendered_diff = diff_markdown(diff)
+    assert "## Process analysis" in rendered_diff
+    assert "slow_customer_search" in rendered_diff
+
+    impact = impact_analysis(old, new)
+    assert impact["analysis_changes"]["pain_points"] == ["slow_customer_search"]
+    assert "check_customer" in impact["analysis_affected_steps"]
+    assert {"outlook", "s4"} <= set(impact["affected"]["systems"])
+    assert "search_screen_01" in impact["affected"]["evidence"]
+    assert any(test["id"] == "check_customer:happy-path" for test in impact["recommended_tests"])
+    rendered_impact = impact_markdown(impact)
+    assert "## Process analysis changes" in rendered_impact
+    assert "Affected steps from analysis" in rendered_impact