#!/usr/bin/env python3 """Bounded fictional workflow search, separate approval and ERP readiness. This is not a Jira workflow interpreter, REST parser or permissions test. The approval edge models the current documented JSM Cloud validator exception; actor eligibility and qualification of the collective final result are separate supplied preconditions. One eligible voter's answer is not the final result. All nodes, edits, consumers and fields below are fictional normalized inputs. No remote operation, workflow change, approval, ERP dispatch, secret or persistent file is involved. """ from collections import deque from dataclasses import dataclass, replace import json import sys STAGES = ("submitted", "awaiting_approval", "approved_awaiting_data", "ready", "exported") KINDS = ("ordinary", "approval", "dispatch") @dataclass(frozen=True) class State: stage: str erp_field_present: bool final_approval_observed: bool edits_used: int @dataclass(frozen=True) class Edge: name: str source: str target: str kind: str required_field_validator: bool = False dispatch_checks_current_data: bool = False @dataclass(frozen=True) class Profile: edges: tuple[Edge, ...] editable_stages: tuple[str, ...] initial_field_present: bool = False actor_qualified_for_final_approval: bool = True final_approval_result_qualified: bool = True inventory_qualified: bool = True max_edits: int = 2 def validate(profile): if (not isinstance(profile.edges, tuple) or not profile.edges or not all(isinstance(e, Edge) for e in profile.edges) or type(profile.max_edits) is not int or not 0 <= profile.max_edits <= 2 or not isinstance(profile.editable_stages, tuple) or any(s not in STAGES[:-1] for s in profile.editable_stages) or len(set(profile.editable_stages)) != len(profile.editable_stages)): raise ValueError("outside_bounded_local_profile") for flag in (profile.initial_field_present, profile.actor_qualified_for_final_approval, profile.final_approval_result_qualified, profile.inventory_qualified): if type(flag) is not bool: raise ValueError("qualification_must_be_boolean") names = set() for e in profile.edges: if (not isinstance(e.name, str) or not e.name.strip() or e.name in names or e.source not in STAGES[:-1] or e.target not in STAGES or e.kind not in KINDS or type(e.required_field_validator) is not bool or type(e.dispatch_checks_current_data) is not bool or ((e.kind == "dispatch") != (e.target == "exported")) or (e.kind != "dispatch" and e.dispatch_checks_current_data)): raise ValueError("edge_outside_local_profile") names.add(e.name) def successors(state, profile): if state.stage == "exported": return [] found = [] for e in profile.edges: if e.source != state.stage: continue if e.kind == "approval": if not (profile.actor_qualified_for_final_approval and profile.final_approval_result_qualified): continue # This exception concerns data validators, not approval permissions. next_state = replace(state, stage=e.target, final_approval_observed=True) else: if e.required_field_validator and not state.erp_field_present: continue if (e.kind == "dispatch" and e.dispatch_checks_current_data and not (state.erp_field_present and state.final_approval_observed)): continue next_state = replace(state, stage=e.target) found.append((e.name, e.kind, next_state)) if state.stage in profile.editable_stages and state.edits_used < profile.max_edits: present = not state.erp_field_present found.append(("fill_erp_field" if present else "clear_erp_field", "edit", replace(state, erp_field_present=present, edits_used=state.edits_used+1))) return found def analyze(profile): validate(profile) start = State("submitted", profile.initial_field_present, False, 0) queue = deque([(start, ())]) seen = {start} approval_witness = None dispatch_witness = None valid_dispatch_witness = None transitions_examined = 0 while queue: state, path = queue.popleft() for name, kind, target in successors(state, profile): transitions_examined += 1 next_path = path+(name,) if kind == "approval" and not target.erp_field_present and approval_witness is None: approval_witness = next_path if kind == "dispatch": complete = target.erp_field_present and target.final_approval_observed if not complete and dispatch_witness is None: dispatch_witness = next_path if complete and valid_dispatch_witness is None: valid_dispatch_witness = next_path if target not in seen: seen.add(target) queue.append((target, next_path)) if dispatch_witness is not None: assessment = "incomplete_dispatch_path_found" elif not profile.inventory_qualified: assessment = "inventory_incomplete_no_clearance" else: assessment = "no_incomplete_dispatch_in_bounded_model" return {"assessment": assessment, "incomplete_approval_path": list(approval_witness) if approval_witness else None, "incomplete_dispatch_path": list(dispatch_witness) if dispatch_witness else None, "complete_dispatch_path": list(valid_dispatch_witness) if valid_dispatch_witness else None, "states_visited": len(seen), "transitions_examined": transitions_examined, "max_edits": profile.max_edits, "native_workflow_certified": False, "distributed_read_and_send_atomicity_certified": False} def run_checks(): names = [] def check(name, condition): if name in names or not condition: raise AssertionError(name) names.append(name) ordinary_entry = Edge("enter_approval", "submitted", "awaiting_approval", "ordinary") approval_to_ready = Edge("final_approval", "awaiting_approval", "ready", "approval", True) unchecked_dispatch = Edge("erp_consumer", "ready", "exported", "dispatch") basic = Profile((ordinary_entry, approval_to_ready, unchecked_dispatch), STAGES[:-1]) guarded = replace(basic, edges=basic.edges[:-1]+(replace(unchecked_dispatch, dispatch_checks_current_data=True),)) before = replace(basic, edges=(replace(ordinary_entry, required_field_validator=True),)+basic.edges[1:]) post_approval = Edge("final_approval", "awaiting_approval", "approved_awaiting_data", "approval", True) ordinary_check = Edge("validate_erp_field", "approved_awaiting_data", "ready", "ordinary", True) after = Profile((ordinary_entry, post_approval, ordinary_check, unchecked_dispatch), STAGES[:-1]) after_guarded = replace(after, edges=after.edges[:-1]+(replace(unchecked_dispatch, dispatch_checks_current_data=True),)) parallel = replace(after_guarded, edges=after_guarded.edges+(Edge("other_erp_consumer", "ready", "exported", "dispatch"),)) unqualified = replace(after_guarded, inventory_qualified=False) no_actor = replace(guarded, actor_qualified_for_final_approval=False) individual_only = replace(guarded, final_approval_result_qualified=False) no_edits_before = replace(before, initial_field_present=True, editable_stages=(), max_edits=0) no_edits_after = replace(after, editable_stages=(), max_edits=0) global_jump = replace(after_guarded, edges=after_guarded.edges+(Edge("global_jump_to_ready", "submitted", "ready", "ordinary"),)) profiles = {"validator_on_approval_only": basic, "sole_connector_guard": guarded, "entry_check_with_later_edits": before, "post_approval_check_with_later_edits": after, "post_check_and_dispatch_guard": after_guarded, "another_unchecked_consumer": parallel, "unqualified_inventory": unqualified, "unqualified_approval_actor": no_actor, "individual_vote_without_final_result": individual_only, "entry_check_without_any_edits": no_edits_before, "post_check_without_field_input": no_edits_after, "global_ready_transition_and_guard": global_jump} results = {name: analyze(p) for name, p in profiles.items()} b = results["validator_on_approval_only"] check("approval_validator_does_not_stop_modeled_approval", b["incomplete_approval_path"] == ["enter_approval", "final_approval"]) check("unguarded_consumer_accepts_incomplete_modeled_dispatch", b["incomplete_dispatch_path"] == ["enter_approval", "final_approval", "erp_consumer"]) check("basic_model_also_has_complete_path", b["complete_dispatch_path"] is not None) g = results["sole_connector_guard"] check("connector_guard_leaves_incomplete_approval_reachable", g["incomplete_approval_path"] is not None) check("sole_guarded_consumer_blocks_incomplete_dispatch", g["incomplete_dispatch_path"] is None) check("sole_guarded_consumer_allows_complete_dispatch", g["complete_dispatch_path"] is not None) be = results["entry_check_with_later_edits"] check("entry_check_can_be_invalidated_by_later_edit", "clear_erp_field" in be["incomplete_approval_path"]) check("entry_check_unguarded_dispatch_still_incomplete", be["incomplete_dispatch_path"] is not None) af = results["post_approval_check_with_later_edits"] check("post_check_accepts_incomplete_approval_by_design", af["incomplete_approval_path"] is not None) check("edit_after_post_check_can_reach_unguarded_consumer", "clear_erp_field" in af["incomplete_dispatch_path"] and af["incomplete_dispatch_path"].index("clear_erp_field") > af["incomplete_dispatch_path"].index("validate_erp_field")) ag = results["post_check_and_dispatch_guard"] check("post_check_and_current_data_guard_block_incomplete_dispatch", ag["incomplete_dispatch_path"] is None) check("post_check_and_guard_still_allow_complete_dispatch", ag["complete_dispatch_path"] is not None) pa = results["another_unchecked_consumer"] check("other_consumer_is_not_protected_by_first_consumer_guard", pa["incomplete_dispatch_path"][-1] == "other_erp_consumer") check("other_consumer_fault_occurs_after_post_check", "validate_erp_field" in pa["incomplete_dispatch_path"]) uq = results["unqualified_inventory"] check("no_witness_does_not_clear_unknown_inventory", uq["incomplete_dispatch_path"] is None and uq["assessment"] == "inventory_incomplete_no_clearance") na = results["unqualified_approval_actor"] check("data_validator_exception_does_not_grant_approval_permission", na["incomplete_approval_path"] is None and na["complete_dispatch_path"] is None) individual = results["individual_vote_without_final_result"] check("eligible_individual_vote_is_not_collective_final_approval", individual_only.actor_qualified_for_final_approval is True and individual["incomplete_approval_path"] is None and individual["complete_dispatch_path"] is None) ne = results["entry_check_without_any_edits"] check("without_edits_entry_check_blocks_incomplete_approval", ne["incomplete_approval_path"] is None) check("without_edits_and_with_data_complete_dispatch_remains_possible", ne["complete_dispatch_path"] is not None) nf = results["post_check_without_field_input"] check("post_check_without_input_blocks_all_dispatch", nf["incomplete_dispatch_path"] is None and nf["complete_dispatch_path"] is None) check("no_dispatch_path_is_not_claimed_as_operational_success", nf["assessment"] == "no_incomplete_dispatch_in_bounded_model" and nf["complete_dispatch_path"] is None) gl = results["global_ready_transition_and_guard"] check("ready_label_alone_does_not_bypass_approval_and_data_guard", gl["incomplete_dispatch_path"] is None) check("guarded_global_transition_does_not_block_legitimate_complete_path", gl["complete_dispatch_path"] is not None) # Vary edge ordering, starting data and edit budget without inventing a # probability or claiming the finite state space is a native workflow. for name, p in profiles.items(): reversed_result = analyze(replace(p, edges=tuple(reversed(p.edges)))) check("edge_order_preserves_reachability_"+name, all((results[name][key] is None) == (reversed_result[key] is None) for key in ("incomplete_approval_path", "incomplete_dispatch_path", "complete_dispatch_path"))) for start_present in (False, True): for budget in (0, 1, 2): p = replace(after_guarded, initial_field_present=start_present, max_edits=budget) value = analyze(p) check("current_data_guard_start_"+str(start_present)+"_budget_"+str(budget), value["incomplete_dispatch_path"] is None) invalid = [replace(basic, max_edits=True), replace(basic, max_edits=3), replace(basic, inventory_qualified=1), replace(basic, actor_qualified_for_final_approval=1), replace(basic, final_approval_result_qualified=1), replace(basic, edges=basic.edges+(ordinary_entry,)), replace(basic, edges=(replace(ordinary_entry, source="unknown"),)+basic.edges[1:]), replace(basic, edges=(replace(ordinary_entry, kind="unknown"),)+basic.edges[1:]), replace(basic, editable_stages=("exported",)), replace(basic, edges=(replace(ordinary_entry, dispatch_checks_current_data=True),)+basic.edges[1:])] for number, p in enumerate(invalid, 1): rejected = False try: analyze(p) except ValueError: rejected = True check("invalid_local_profile_"+str(number), rejected) check("all_searches_explore_real_local_states", all(r["states_visited"] > 0 and r["transitions_examined"] >= 0 for r in results.values())) check("all_terminal_export_states_have_no_later_model_edits", successors(State("exported", True, True, 0), basic) == []) check("input_profile_is_unchanged", basic.edges == (ordinary_entry, approval_to_ready, unchecked_dispatch) and basic.max_edits == 2) return {"checks_passed": len(names), "check_names": names, "python_version": sys.version, "profile_scenarios": results, "scope": {"http_calls": 0, "jira_mutations": 0, "approvals_performed": 0, "erp_operations": 0, "persistent_writes": 0, "native_workflow_interpreter": False, "native_permissions_certified": False, "distributed_atomicity_certified": False, "maximum_field_edits_per_path": 2, "search_is_actual_breadth_first_state_exploration": True}} if __name__ == "__main__": print(json.dumps(run_checks(), ensure_ascii=False, indent=2))