Merge pull request #29 from StrategyLogic:release/action-condition

Simplify workflow and improve action conditions

Author: RCFans

.github/workflows/release.yml

@@ -23,10 +23,9 @@ jobs:
         python-version: "3.12"
     - name: Install build dependencies
       run: |
-        python -m pip install --upgrade pip
-        python -m pip install -e .
-        python -m pip install -e ".[dev]"
-        python -m pip install twine
+        pip install --upgrade pip
+        pip install -e .
+        pip install twine
     - name: Build distribution artifacts
       run: |
         python -m build --sdist --wheel

README.md

@@ -6,7 +6,7 @@
 ![Codecov](https://img.shields.io/codecov/c/github/StrategyLogic/omen) [![Package](https://img.shields.io/github/actions/workflow/status/StrategyLogic/omen/package.yml)](https://img.shields.io/github/actions/workflow/status/StrategyLogic/omen/package.yml) ![License](https://img.shields.io/pypi/l/omenai) ![Downloads](https://img.shields.io/pepy/dt/omenai) ![PyPI Version](https://img.shields.io/pypi/v/omenai)
 </div>
 
-[**Omen**](https://github.com/StrategyLogic/omen) (Chinese: 爻) is a strategic reasoning engine based on **Explainable AI**. It leverages **ontological modeling** to understand the phenomena and essence of the strategic world, and **counterfactual analysis** to simulate the known and unknown in decision-making scenarios, generating verifiable, comparable, and explainable decision support for decision-makers.
+[**Omen**](https://github.com/StrategyLogic/omen) (Chinese: 爻) is a strategic reasoning engine based on **Explainable AI**. It leverages **ontological modeling** to understand the phenomena and essence of the strategic world, and **counterfactual analysis** to simulate the known and unknown in decision-making scenarios, generating verifiable, traceable, and explainable strategic insights for decision-makers.
 
 [Concepts](docs/concepts.md) | [Quick Start](docs/quick-start.md) |  [Case Templates](docs/case-template.md) | [Roadmap](docs/roadmap.md)
 

README.zh.md

@@ -7,7 +7,7 @@
 
 </div>
 
-[**Omen**](https://github.com/StrategyLogic/omen) （中文：爻）是基于**可解释AI**的战略推演引擎。它以**本体论建模**理解战略世界中的现象与本质，以**反事实分析**模拟决策情景中的已知与未知，为决策者生成可验证、可比较和可解释的决策依据。
+[**Omen**](https://github.com/StrategyLogic/omen) （中文：爻）是基于**可解释AI**的战略推演引擎。它以**本体论建模**理解战略世界中的现象与本质，以**反事实分析**模拟决策情景中的已知与未知，为决策者生成可验证、可追溯和可解释的战略洞察。
 
 [核心概念](docs/concepts.md) | [快速开始](docs/quick-start.md) | [案例模板](docs/case-template.md) | [项目路线图](docs/roadmap.md)
 

src/omen/analysis/actor/derivation_trace.py

@@ -98,13 +98,37 @@ def _normalize_items(items: list[Any], *, bucket: str) -> list[dict[str, Any]]:
 
 def build_linked_evidence_refs(reason_chain: dict[str, Any]) -> list[dict[str, Any]]:
     conclusions = extract_conclusion_buckets(dict(reason_chain.get("conclusions") or {}))
+    steps = [item for item in (reason_chain.get("steps") or []) if isinstance(item, dict)]
+    default_reason_step_ids = [
+        str(item.get("step_id") or "").strip()
+        for item in steps
+        if str(item.get("step_type") or "").strip()
+        in {"required_or_warning_or_blocking", "strategic_conditions"}
+        and str(item.get("step_id") or "").strip()
+    ]
+    default_activation_step_ids = [
+        str(item.get("step_id") or "").strip()
+        for item in steps
+        if str(item.get("step_type") or "").strip() == "constraint_activation"
+        and str(item.get("step_id") or "").strip()
+    ]
+
+    if not default_reason_step_ids:
+        fallback = str(steps[-1].get("step_id") or "").strip() if steps else ""
+        if fallback:
+            default_reason_step_ids = [fallback]
+
     refs: list[dict[str, Any]] = []
     for bucket in ("required", "warning", "blocking"):
         for index, item in enumerate(list(conclusions.get(bucket) or []), start=1):
             if not isinstance(item, dict):
                 continue
             reason_ids = [str(x).strip() for x in (item.get("reason_step_ids") or []) if str(x).strip()]
             activation_ids = [str(x).strip() for x in (item.get("activation_step_ids") or []) if str(x).strip()]
+            if not reason_ids:
+                reason_ids = list(default_reason_step_ids)
+            if bucket == "blocking" and not activation_ids:
+                activation_ids = list(default_activation_step_ids)
             refs.append(
                 {
                     "evidence_id": f"{bucket}_{index}",
@@ -298,9 +322,11 @@ def build_reason_chain_view_model_artifact(
         scenario_key = str(row.get("scenario_key") or "")
         reason_chain = dict(row.get("reason_chain") or {})
         steps = list(reason_chain.get("steps") or [])
+        ordered_step_node_ids: list[str] = []
         for step in steps:
             step_id = str(step.get("step_id") or "").strip()
             node_id = f"{scenario_key}:{step_id}"
+            ordered_step_node_ids.append(node_id)
             graph_nodes.append(
                 {
                     "id": node_id,
@@ -312,6 +338,16 @@ def build_reason_chain_view_model_artifact(
                 }
             )
 
+        # Keep DAG connected in workshop mode even when conclusions are empty.
+        for index in range(1, len(ordered_step_node_ids)):
+            graph_edges.append(
+                {
+                    "from": ordered_step_node_ids[index - 1],
+                    "to": ordered_step_node_ids[index],
+                    "edge_type": "next",
+                }
+            )
+
         conclusions = dict(reason_chain.get("conclusions") or {})
         for bucket in ("required", "warning", "blocking"):
             for index, item in enumerate(list(conclusions.get(bucket) or []), start=1):

src/omen/cli/case.py

@@ -10,7 +10,7 @@
 
 from omen.analysis.actor.derivation import (
     derive_actor_path,
-    derive_strategic_freedom_conditions,
+    get_simulate_reasoning_order,
 )
 from omen.analysis.actor.derivation_trace import (
     extract_conclusion_buckets,
@@ -146,17 +146,21 @@ def run_deterministic_simulate_from_pack(
             capability_scores=capability_templates.get(scenario_key, {}),
             capability_fit=capability_fit["fit"],
         )
-        strategic_conditions = derive_strategic_freedom_conditions(
-            scenario_key=scenario_key,
-            actor_derivation=actor_derivation,
-            selected_dimensions=selected_dimensions,
-            resistance_baseline=scenario["resistance_baseline"],
-            capability_fit=capability_fit["fit"],
-        )
         confidence_level, missing_reasons = apply_partial_evidence_confidence_policy(
             evidence_refs=[],
             scenario_key=scenario_key,
         )
+        strategic_score = calculate_strategic_freedom_factor(
+            capability_fit=capability_fit["fit"],
+            resistance_baseline=scenario["resistance_baseline"],
+        )
+        strategic_conditions = {
+            "score": strategic_score,
+            "required": [],
+            "warning": [],
+            "blocking": [],
+            "reasoning_order": list(get_simulate_reasoning_order()),
+        }
         derivation_trace = build_actor_derivation_trace(
             scenario_key=scenario_key,
             scenario_ontology=scene,
@@ -165,10 +169,6 @@ def run_deterministic_simulate_from_pack(
             strategic_conditions=strategic_conditions,
             missing_evidence_reasons=missing_reasons,
         )
-        strategic_score = calculate_strategic_freedom_factor(
-            capability_fit=capability_fit["fit"],
-            resistance_baseline=scenario["resistance_baseline"],
-        )
 
         scenario_results.append(
             {
@@ -246,7 +246,12 @@ def run_deterministic_simulate_from_pack(
             }
             llm_payload = try_generate_scenario_reason_chain_via_llm(
                 scenario_json=llm_scenario_input,
-                actor_profile_json={"actor_profile_ref": actor_profile_ref},
+                actor_profile_json={
+                    "actor_profile_ref": actor_profile_ref,
+                    "actor_derivation": dict(result.get("actor_derivation") or {}),
+                    "selected_dimensions": list((result.get("selected_dimensions") or {}).get("selected_dimension_keys") or []),
+                    "resistance": dict(result.get("resistance") or {}),
+                },
                 planning_query_json={},
                 situation_markdown="",
                 config_path=config_path,
@@ -318,12 +323,11 @@ def run_deterministic_simulate_from_pack(
             key = str(result.get("scenario_key") or "")
             reason_chain = chain_by_key.get(key, {})
             buckets = extract_conclusion_buckets(dict(reason_chain.get("conclusions") or {}))
-            if any(buckets[name] for name in ("required", "warning", "blocking")):
-                freedom = dict(result.get("strategic_freedom") or {})
-                freedom["required"] = [str(item.get("text") or "").strip() for item in buckets["required"] if str(item.get("text") or "").strip()]
-                freedom["warning"] = [str(item.get("text") or "").strip() for item in buckets["warning"] if str(item.get("text") or "").strip()]
-                freedom["blocking"] = [str(item.get("text") or "").strip() for item in buckets["blocking"] if str(item.get("text") or "").strip()]
-                result["strategic_freedom"] = freedom
+            freedom = dict(result.get("strategic_freedom") or {})
+            freedom["required"] = [str(item.get("text") or "").strip() for item in buckets["required"] if str(item.get("text") or "").strip()]
+            freedom["warning"] = [str(item.get("text") or "").strip() for item in buckets["warning"] if str(item.get("text") or "").strip()]
+            freedom["blocking"] = [str(item.get("text") or "").strip() for item in buckets["blocking"] if str(item.get("text") or "").strip()]
+            result["strategic_freedom"] = freedom
             result["evidence_refs"] = build_linked_evidence_refs(reason_chain)
 
             if result["evidence_refs"]:

tests/contract/test_reason_chain_view_model_contract.py

@@ -114,4 +114,3 @@ def test_reason_chain_view_model_contract_in_workshop_mode(tmp_path: Path, monke
     assert nodes
     assert edges
     assert any(str(node.get("node_type") or "") == "reason_step" for node in nodes)
-    assert any(str(node.get("node_type") or "") == "claim" for node in nodes)

tests/integration/test_reason_chain_evidence_refs.py

@@ -102,8 +102,7 @@ def test_reason_chain_evidence_refs_link_to_reason_steps(tmp_path: Path, monkeyp
     assert rows
     for row in rows:
         evidence_refs = list(row.get("evidence_refs") or [])
-        assert evidence_refs
-        assert all(list(item.get("reason_step_ids") or []) for item in evidence_refs)
-        for item in evidence_refs:
-            if item.get("bucket") == "blocking":
-                assert list(item.get("activation_step_ids") or [])
+        assert not evidence_refs
+        assert str(row.get("confidence_level") or "") == "reduced-confidence"
+        missing = list((row.get("derivation_trace") or {}).get("missing_evidence_reasons") or [])
+        assert missing

tests/unit/test_reason_chain_evidence_ref_backfill.py

@@ -0,0 +1,34 @@
+from omen.analysis.actor.derivation_trace import build_linked_evidence_refs
+
+
+def test_evidence_refs_backfill_reason_step_ids_from_reason_chain_steps() -> None:
+  reason_chain = {
+    "steps": [
+      {"step_id": "step_1", "step_type": "seed"},
+      {"step_id": "step_2", "step_type": "constraint_activation"},
+      {"step_id": "step_5", "step_type": "required_or_warning_or_blocking"},
+    ],
+    "conclusions": {
+      "required": [
+        {"text": "Must unblock alliance dependency"}
+      ],
+      "warning": [
+        {"text": "Risk of migration delay"}
+      ],
+      "blocking": [
+        {"text": "Cannot proceed under veto pressure"}
+      ],
+    },
+  }
+
+  refs = build_linked_evidence_refs(reason_chain)
+  assert refs
+
+  required = next(item for item in refs if item.get("bucket") == "required")
+  warning = next(item for item in refs if item.get("bucket") == "warning")
+  blocking = next(item for item in refs if item.get("bucket") == "blocking")
+
+  assert required.get("reason_step_ids") == ["step_5"]
+  assert warning.get("reason_step_ids") == ["step_5"]
+  assert blocking.get("reason_step_ids") == ["step_5"]
+  assert blocking.get("activation_step_ids") == ["step_2"]