Add deterministic process graph liveness validation

docs/architecture.md

@@ -6,6 +6,9 @@ Process as Code is a deterministic contract tool, not a workflow runtime.
 YAML/JSON contract
   -> loader
   -> structural + graph/reference validator
+     -> reachable graph
+     -> reverse terminal reachability
+     -> trapped-cycle / gateway diagnostics
   -> semantic model by stable IDs
      -> Mermaid / BPMN / Markdown / RACI
      -> semantic & visual diff
@@ -21,4 +24,8 @@ YAML/JSON contract
 
 The v0.2 JSON Schema describes syntax; Python validation enforces graph and cross-reference invariants. Stable IDs are the semantic backbone for diff, impact, tests, external links and knowledge-graph export.
 
-Vendor-specific information belongs under extensions. Optional integrations (MCP, editor tooling, external adapters) sit outside the deterministic core and must not change the meaning of a valid contract.
+Graph validation deliberately separates **reachability** from **liveness**. Forward traversal identifies what execution can enter. Reachable edge-less steps define the terminal set. A reverse traversal then determines which reachable steps can eventually reach a terminal. Strongly connected components identify trapped retry/rework cycles that have no terminating path without treating every intentional loop as an error.
+
+The validator remains a static contract analyzer: a liveness-clean graph means the declared model contains a terminating path from every reachable branch, not that external workers or systems are guaranteed to complete it.
+
+Vendor-specific information belongs under extensions. Optional integrations (MCP, editor tooling, external adapters) sit outside the deterministic core and must not change the meaning of a valid contract.

docs/specification.md

@@ -1,6 +1,6 @@
 # Process Contract Specification v0.2
 
-This document is normative for the concepts described here. `schemas/process.schema.json` is the machine-readable schema. The Python validator additionally checks graph reachability and cross-reference integrity that JSON Schema alone cannot express conveniently.
+This document is normative for the concepts described here. `schemas/process.schema.json` is the machine-readable schema. The Python validator additionally checks graph reachability, liveness, gateway semantics and cross-reference integrity that JSON Schema alone cannot express conveniently.
 
 ## Required top-level fields
 
@@ -40,6 +40,24 @@ transitions:
 
 Legacy `next` and `branches` remain accepted by the reference validator for migration compatibility, but `process-code migrate` converts them to v0.2 transitions.
 
+## Graph liveness
+
+A structurally connected process is not necessarily able to finish. The reference validator therefore evaluates liveness on the graph reachable from `process.start`.
+
+A **terminal step** is a reachable step with no outgoing transition. `type: end` is the explicit terminal form and must not declare outgoing transitions. For v0.1 compatibility a reachable non-`end` step with no outgoing transition remains valid, but it is reported as an **implicit terminal** warning.
+
+The validator computes the reverse reachability set from all reachable terminals. Every reachable step outside that set is reported because execution can enter it but can never reach a terminal. An unreachable terminal does not make an otherwise trapped process live.
+
+Cycles are allowed. A retry/rework loop is valid when at least one path can leave the loop and eventually reach a terminal. A strongly connected cycle component whose reachable nodes cannot reach any terminal is reported separately as a **trapped cycle** so it is distinguishable from ordinary loop semantics.
+
+Gateway guidance is deliberately conservative:
+
+- a `decision` requires at least one outgoing edge structurally and warns when it has fewer than two branches;
+- a `parallel` step warns when it has neither multiple incoming nor multiple outgoing flows, because it is then not acting as a meaningful split or join;
+- these checks describe graph semantics only; the core does not execute conditions or parallel runtime behavior.
+
+Liveness findings are deterministic model-quality diagnostics. They do not claim that external systems, humans or runtime workers will actually complete the business process.
+
 ## Inputs and outputs
 
 Each contract can declare `id`, `name`, `type`, `ref`, and `required`. At least one of `id`, `name`, or `ref` must exist.
@@ -54,4 +72,4 @@ Risks are catalog entities and may include a severity such as `low`, `medium`, `
 
 ## Extensions
 
-Unknown fields are allowed so vendor/domain extensions can evolve independently. Namespaced extension objects are recommended, for example `extensions.sap`.
+Unknown fields are allowed so vendor/domain extensions can evolve independently. Namespaced extension objects are recommended, for example `extensions.sap`.

examples/adapters/process-manifest.csv

@@ -1,4 +1,4 @@
 process_id,process_name,step_id,step_name,type,actor,actor_name,system,system_name,next
 simple_order,Simple Order,request,Request order,user_task,sales,Sales,erp,ERP,approve
-simple_order,Simple Order,approve,Approve order,decision,manager,Manager,erp,ERP,complete
-simple_order,Simple Order,complete,Complete,end,,,,,
+simple_order,Simple Order,approve,Approve order,user_task,manager,Manager,erp,ERP,complete
+simple_order,Simple Order,complete,Complete,end,,,,,

src/process_as_code/graph.py

@@ -61,6 +61,95 @@ def reachable_step_ids(process: dict[str, Any]) -> set[str]:
     return seen
 
 
+def terminal_step_ids(process: dict[str, Any], *, reachable_only: bool = False) -> set[str]:
+    """Return steps with no outgoing graph edges.
+
+    A non-`end` step can still be an implicit terminal for v0.1 compatibility. The
+    validator reports that shape separately; this helper only models graph liveness.
+    """
+    graph = adjacency(process)
+    terminals = {node for node, edges in graph.items() if not edges}
+    if reachable_only:
+        terminals &= reachable_step_ids(process)
+    return terminals
+
+
+def steps_reaching_any(process: dict[str, Any], targets: set[str]) -> set[str]:
+    """Return nodes that can reach at least one target, including targets themselves."""
+    graph = adjacency(process)
+    reverse: dict[str, set[str]] = defaultdict(set)
+    for source, edges in graph.items():
+        for target, _ in edges:
+            if target in graph:
+                reverse[target].add(source)
+
+    seen: set[str] = set()
+    queue = deque(sorted(targets & set(graph)))
+    while queue:
+        node = queue.popleft()
+        if node in seen:
+            continue
+        seen.add(node)
+        for source in sorted(reverse.get(node, set())):
+            if source not in seen:
+                queue.append(source)
+    return seen
+
+
+def strongly_connected_components(process: dict[str, Any], nodes: set[str] | None = None) -> list[set[str]]:
+    """Return deterministic strongly connected components for the selected graph nodes."""
+    graph = adjacency(process)
+    allowed = set(graph) if nodes is None else set(graph) & nodes
+    index = 0
+    indices: dict[str, int] = {}
+    lowlinks: dict[str, int] = {}
+    stack: list[str] = []
+    on_stack: set[str] = set()
+    components: list[set[str]] = []
+
+    def visit(node: str) -> None:
+        nonlocal index
+        indices[node] = index
+        lowlinks[node] = index
+        index += 1
+        stack.append(node)
+        on_stack.add(node)
+
+        for target, _ in graph.get(node, []):
+            if target not in allowed:
+                continue
+            if target not in indices:
+                visit(target)
+                lowlinks[node] = min(lowlinks[node], lowlinks[target])
+            elif target in on_stack:
+                lowlinks[node] = min(lowlinks[node], indices[target])
+
+        if lowlinks[node] != indices[node]:
+            return
+        component: set[str] = set()
+        while stack:
+            member = stack.pop()
+            on_stack.remove(member)
+            component.add(member)
+            if member == node:
+                break
+        components.append(component)
+
+    for node in sorted(allowed):
+        if node not in indices:
+            visit(node)
+    return sorted(components, key=lambda component: tuple(sorted(component)))
+
+
+def is_cycle_component(process: dict[str, Any], component: set[str]) -> bool:
+    if len(component) > 1:
+        return True
+    if not component:
+        return False
+    node = next(iter(component))
+    return any(target == node for target, _ in adjacency(process).get(node, []))
+
+
 def incoming_counts(process: dict[str, Any]) -> dict[str, int]:
     counts: dict[str, int] = defaultdict(int)
     for edges in adjacency(process).values():
@@ -72,4 +161,4 @@ def incoming_counts(process: dict[str, Any]) -> dict[str, int]:
 def iter_entity_ids(process: dict[str, Any], section: str) -> Iterable[str]:
     for item in process.get(section, []) or []:
         if isinstance(item, dict) and isinstance(item.get("id"), str):
-            yield item["id"]
+            yield item["id"]

src/process_as_code/validate.py

@@ -3,7 +3,16 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-from .graph import iter_entity_ids, reachable_step_ids, step_edges
+from .graph import (
+    incoming_counts,
+    is_cycle_component,
+    iter_entity_ids,
+    reachable_step_ids,
+    step_edges,
+    steps_reaching_any,
+    strongly_connected_components,
+    terminal_step_ids,
+)
 
 
 @dataclass
@@ -118,7 +127,8 @@ def validate_process(data: dict[str, Any]) -> ValidationResult:
         if not isinstance(step, dict) or not isinstance(step.get("id"), str):
             continue
         sid = step["id"]
-        for target, _ in step_edges(step):
+        edges = step_edges(step)
+        for target, _ in edges:
             if target not in step_id_set:
                 result.errors.append(f"step '{sid}' references unknown next step '{target}'")
         transitions = step.get("transitions")
@@ -128,8 +138,13 @@ def validate_process(data: dict[str, Any]) -> ValidationResult:
             for index, transition in enumerate(transitions):
                 if not isinstance(transition, dict) or not isinstance(transition.get("to"), str):
                     result.errors.append(f"step '{sid}' transitions[{index}] requires string 'to'")
-        if step.get("type") == "decision" and not step_edges(step):
-            result.errors.append(f"decision step '{sid}' requires transitions or branches")
+        if step.get("type") == "end" and edges:
+            result.errors.append(f"end step '{sid}' must not declare outgoing transitions")
+        if step.get("type") == "decision":
+            if not edges:
+                result.errors.append(f"decision step '{sid}' requires transitions or branches")
+            elif len(edges) < 2:
+                result.warnings.append(f"decision step '{sid}' has fewer than two outgoing branches")
         if step.get("type") == "subprocess" and not isinstance(step.get("process_ref"), str):
             result.errors.append(f"subprocess step '{sid}' requires process_ref")
 
@@ -155,7 +170,33 @@ def validate_process(data: dict[str, Any]) -> ValidationResult:
         reachable = reachable_step_ids(data)
         for sid in sorted(step_id_set - reachable):
             result.warnings.append(f"step '{sid}' is unreachable from process start")
-        if not any(not step_edges(step) for step in steps if isinstance(step, dict)):
-            result.warnings.append("process has no terminal step")
 
-    return result
+        by_id = {
+            step["id"]: step
+            for step in steps
+            if isinstance(step, dict) and isinstance(step.get("id"), str)
+        }
+        incoming = incoming_counts(data)
+        for sid in sorted(reachable):
+            step = by_id[sid]
+            edges = step_edges(step)
+            if step.get("type") != "end" and not edges:
+                result.warnings.append(f"non-end step '{sid}' is an implicit terminal with no outgoing transition")
+            if step.get("type") == "parallel" and incoming.get(sid, 0) < 2 and len(edges) < 2:
+                result.warnings.append(
+                    f"parallel step '{sid}' has neither multiple incoming nor multiple outgoing flows"
+                )
+
+        terminals = terminal_step_ids(data, reachable_only=True)
+        if not terminals:
+            result.warnings.append("process has no reachable terminal step")
+        can_reach_terminal = steps_reaching_any(data, terminals)
+        for sid in sorted(reachable - can_reach_terminal):
+            result.warnings.append(f"reachable step '{sid}' has no path to a terminal step")
+
+        for component in strongly_connected_components(data, reachable):
+            if is_cycle_component(data, component) and component.isdisjoint(can_reach_terminal):
+                members = ", ".join(sorted(component))
+                result.warnings.append(f"trapped cycle component has no path to a terminal step: {members}")
+
+    return result

tests/test_liveness.py

@@ -0,0 +1,114 @@
+from process_as_code.validate import validate_process
+
+
+def _process(steps, start="start"):
+    return {
+        "version": "0.2",
+        "process": {"id": "liveness", "name": "Liveness", "start": start},
+        "steps": steps,
+    }
+
+
+def _step(step_id, step_type="task", transitions=None):
+    step = {"id": step_id, "name": step_id.replace("_", " ").title(), "type": step_type}
+    if transitions is not None:
+        step["transitions"] = [{"to": target} for target in transitions]
+    return step
+
+
+def test_terminating_retry_loop_is_not_reported_as_trapped():
+    data = _process(
+        [
+            _step("start", transitions=["gate"]),
+            _step("gate", "decision", ["retry", "done"]),
+            _step("retry", transitions=["gate"]),
+            _step("done", "end"),
+        ]
+    )
+
+    result = validate_process(data)
+
+    assert result.ok
+    assert result.warnings == []
+
+
+def test_reachable_trapped_cycle_is_distinct_from_unreachable_terminal():
+    data = _process(
+        [
+            _step("start", transitions=["retry_a"]),
+            _step("retry_a", transitions=["retry_b"]),
+            _step("retry_b", transitions=["retry_a"]),
+            _step("unused_end", "end"),
+        ]
+    )
+
+    result = validate_process(data)
+
+    assert result.ok
+    assert "step 'unused_end' is unreachable from process start" in result.warnings
+    assert "process has no reachable terminal step" in result.warnings
+    assert "reachable step 'start' has no path to a terminal step" in result.warnings
+    assert "reachable step 'retry_a' has no path to a terminal step" in result.warnings
+    assert "reachable step 'retry_b' has no path to a terminal step" in result.warnings
+    assert "trapped cycle component has no path to a terminal step: retry_a, retry_b" in result.warnings
+
+
+def test_end_step_with_outgoing_transition_is_error():
+    data = _process(
+        [
+            _step("start", "end", ["after"]),
+            _step("after", "end"),
+        ]
+    )
+
+    result = validate_process(data)
+
+    assert not result.ok
+    assert "end step 'start' must not declare outgoing transitions" in result.errors
+
+
+def test_reachable_non_end_without_outgoing_transition_is_implicit_terminal_warning():
+    result = validate_process(_process([_step("start")]))
+
+    assert result.ok
+    assert result.warnings == ["non-end step 'start' is an implicit terminal with no outgoing transition"]
+
+
+def test_decision_with_only_one_branch_warns():
+    data = _process(
+        [
+            _step("start", "decision", ["done"]),
+            _step("done", "end"),
+        ]
+    )
+
+    result = validate_process(data)
+
+    assert result.ok
+    assert "decision step 'start' has fewer than two outgoing branches" in result.warnings
+
+
+def test_parallel_one_in_one_out_warns_but_split_and_join_do_not():
+    weak = _process(
+        [
+            _step("start", transitions=["parallel"]),
+            _step("parallel", "parallel", ["done"]),
+            _step("done", "end"),
+        ]
+    )
+    weak_result = validate_process(weak)
+    assert "parallel step 'parallel' has neither multiple incoming nor multiple outgoing flows" in weak_result.warnings
+
+    valid = _process(
+        [
+            _step("start", transitions=["split"]),
+            _step("split", "parallel", ["left", "right"]),
+            _step("left", transitions=["join"]),
+            _step("right", transitions=["join"]),
+            _step("join", "parallel", ["done"]),
+            _step("done", "end"),
+        ]
+    )
+    valid_result = validate_process(valid)
+    assert valid_result.ok
+    assert not [warning for warning in valid_result.warnings if "parallel step" in warning]