Skip to content

mineproductivity.decision

Auto-generated API reference for the Decision package, rendered from the package's own docstrings. For the narrative overview, dependency rules, and extension guide, see Packages -> Decision.

decision

mineproductivity.decision -- the platform's prescriptive layer, built directly on top of analytics.

Answers the question analytics deliberately does not: given a trend, a benchmark classification, a baseline, or a confidence interval, what should the business actually do about it? decision consumes already-correct kpis.KPIResult/analytics.AnalyticsResult evidence and produces recommended, ranked, explained, actionable decisions as new, equally discoverable, equally versioned result objects -- never recomputing a KPI value, a trend, or a benchmark classification itself.

Phase 07.1 -- Decision Intelligence Foundation implements every component required before concrete decision models can exist: DecisionModel (ABC)/DecisionContext (§8); DecisionMetadata/ DecisionCategory (§30); the full DecisionResult family (§28); Threshold (§13); DecisionPipeline/PipelineStage/ModelStage (§9); REGISTRY/register (§32); and the full exception hierarchy.

Phase 07.2 -- Decision Rule Engine implements rule evaluation, policy governance, and the default decision strategy: Rule/ RuleEngine/RuleEngineStage (§10); Policy/DecisionStatus (§12); DecisionStrategy (ABC)/ThresholdDecisionStrategy (§14), self-registered into REGISTRY at import time.

Phase 07.3 -- Decision Intelligence Analysis Layer implements the analytical capabilities that operate on already-produced Recommendation/DecisionResult objects: DecisionScorer/ ConfidenceScorer (§23, §24) -- the numeric weight backing ranking and the trust weight backing how strongly a recommendation should be acted on, the latter deriving from analytics.DataQualityScore when present in a DecisionContext's evidence, never recomputing data quality itself; RankingStrategy (ABC)/WeightedScoreRanking (§16) -- orders Recommendation\ s by DecisionScore.value, descending, self-registered into REGISTRY; ExplanationBuilder/ ExplanationStage (§17) -- walks a Recommendation's provenance into a structured, evidence-linked Explanation; ActionPrioritizer (§20) -- assigns each RankedRecommendation an ActionPriority (urgency/impact/effort), a distinct question from ranking; and RootCauseAnalyzer (§18) -- an interface-only ABC with zero concrete subclasses by design (ADR-0007), defining the contract a future causal- inference plugin implements. DecisionContext gained a second, Phase-07.3 extension (recommendations, alongside Phase 07.2's triggered_rules) for the same reason: WeightedScoreRanking/ ExplanationStage operate over a batch of already-produced Recommendation\ s, not raw evidence (see abstractions.DecisionContext's own docstring).

Phase 07.4 -- Decision Operational Services completes the package: WhatIfEngine (§19) -- an interface-only ABC with zero concrete subclasses by design, the decision-layer counterpart to RootCauseAnalyzer, reusing events.AsOf's already-reserved scenario field rather than inventing a second scenario concept; ActionPlanner (§21) -- sequences prioritized actions into an ActionPlan, respecting declared dependencies via its own narrow, self-contained topological ordering (deliberately not kpis.DependencyGraph, per §33's own anti-pattern entry); AlertGenerator (§22) -- produces an Alert from a ThresholdBreach or a high-severity Recommendation, a pure value- object factory with no channel-delivery side effects; RealTimeDecisionSession/BatchDecisionRunner (§25, §26) -- the two execution modes, the live event-driven counterpart and the bounded, scheduled-report counterpart of running one DecisionPipeline, both composing kpis.KPIEngine/analytics.BatchAnalyticsRunner rather than recomputing anything themselves; and DecisionAuditTrail/ DecisionAuditEntry (§27, §28) -- the append-only accountability record every operationally-actionable DecisionPipeline run should feed, optionally wired into both execution modes. recommendation.py (§15, §6) holds the recommendation-generation logic with no public API of its own, per §6's own entry for it -- ThresholdDecisionStrategy's methods delegate Recommendation construction to it, keeping exactly one summary-text/traceability format package-wide. This completes every module design spec §6 enumerates -- decision is now feature-complete per the Reference Implementation Blueprint. See the package's own README.md for the full slice inventory and phase status.

decision depends on core, events, ontology, registry, plugins, connectors, kpis, and analytics -- and MUST NEVER import digital_twin, simulation, optimization, visualization, or agents, none of which this package may see.

Everything documented here is part of the public API and can be imported directly from mineproductivity.decision, e.g.::

from mineproductivity.decision import DecisionModel, DecisionContext

Rule

Rule = BaseSpecification[DecisionContext]

Reuses core.BaseSpecification (&/|/~ composable) exactly as events.EventFilter already does -- rule evaluation is never a bespoke DSL in this platform.

DecisionContext

DecisionContext(*, kpi_results, analytics_results, scope, event_store=None, as_of=None, triggered_rules=None, recommendations=None)

Bundles the collaborators and scope a DecisionModel needs -- the decision-layer counterpart to analytics.AnalyticsContext, one layer up. Carries the KPIResult/AnalyticsResult evidence already gathered, the scope the decision concerns (e.g. a specific pit or shift), and optionally an EventStore/AsOf framing a point-in-time or hypothetical-scenario view for future root-cause/ what-if use.

triggered_rules is populated by :class:~mineproductivity.decision.rules.RuleEngineStage (design spec §10) as a pipeline runs, so a downstream :class:~mineproductivity.decision.strategy.ThresholdDecisionStrategy never has to re-run rule evaluation itself -- a necessary, minimal, disclosed extension of this class beyond design spec §8's own pseudocode (which shows no such field), added in Phase 07.2 because §10's RuleEngineStage explicitly needs some way to pass its result to the next stage, and this class is the only vehicle a PipelineStage.process has for doing so. Defaults to None, meaning "no RuleEngineStage has run yet" -- deliberately not (), which is instead the legitimate, common result of a RuleEngineStage that ran and found nothing triggered. Collapsing both into the same falsy default would make "not yet computed" and "computed as empty" indistinguishable to a downstream consumer (caught and fixed during this phase's own QA review, before it could silently defeat the "never re-run" guarantee for precisely the most common case).

recommendations is a Phase 07.3 extension of the same shape and for the same reason as triggered_rules: design spec §16's WeightedScoreRanking and §17's ExplanationStage are, per their own pseudocode, DecisionModel/PipelineStage implementations that operate over a batch of already-produced Recommendation\ s, not over raw KPIResult/AnalyticsResult evidence -- yet _decide/process receive only a DecisionContext, the one vehicle available for a caller (or an upstream, non-terminal pipeline stage) to hand a batch across. Also defaults to None for consistency with triggered_rules's own field shape -- though, unlike triggered_rules, no consumer in this phase currently branches differently on None versus () (:class:~mineproductivity.decision.ranking.WeightedScoreRanking and :class:~mineproductivity.decision.explanation.ExplanationStage both treat "not supplied" and "supplied, empty" identically, via a plain falsy check); the distinction is preserved as a matter of shape consistency and to leave room for a future consumer that does need to tell the two apart, not because anything here relies on it today.

Examples:

>>> ctx = DecisionContext(kpi_results=(), analytics_results=(), scope={"pit": "north"})
>>> ctx.scope
{'pit': 'north'}
>>> ctx.event_store is None
True
>>> ctx.triggered_rules is None
True
>>> ctx.recommendations is None
True

DecisionModel

Bases: ABC

The root of every registrable decision strategy -- 'Decision-as- object,' the direct counterpart of kpis.BaseKPI and analytics.AnalyticsModel, one and two layers down respectively. A concrete leaf declares meta: ClassVar[DecisionMetadata] and implements :meth:_decide; everything else (input validation, result envelope wrapping) is inherited.

Thread safety. Every DecisionModel subclass MUST be stateless across :meth:decide calls -- no instance attribute is mutated by :meth:_decide -- so a single instance is safe to share and invoke concurrently from multiple threads.

decide

decide(context)

Non-overridden orchestration: validates context carries at least one KPIResult or AnalyticsResult to reason over, then calls :meth:_decide.

AlertGenerator

AlertGenerator(*, breach_severity='high')

Produces an Alert from a ThresholdBreach (§13) or a high-severity Recommendation (§15). A concrete, non-pluggable utility -- alert channel delivery (email, SMS, a dashboard push) is explicitly out of scope for this package (§4); AlertGenerator produces the Alert value object only, never sends it anywhere, keeping it a pure, easily-tested function with no I/O side effects.

Examples:

>>> from datetime import datetime, timezone
>>> from mineproductivity.decision.thresholds import Threshold
>>> breach = ThresholdBreach(
...     threshold=Threshold(field="value", comparator="<", limit=0.65),
...     observed_value=0.58, breached_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... )
>>> AlertGenerator().from_breach(breach).severity
'high'
>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="Investigate fleet availability", severity="critical", evidence=("UTIL.OEE",),
... )
>>> AlertGenerator().from_recommendation(rec).message
'Investigate fleet availability'
>>> low_severity = Recommendation(
...     policy_code="X", triggered_rules=(), summary="x", severity="low", evidence=(),
... )
>>> AlertGenerator().from_recommendation(low_severity) is None
True

DecisionAuditEntry dataclass

DecisionAuditEntry(recorded_at, result, context_scope, *, source_event_ids=())

Bases: BaseValueObject

One append-only record: what was decided, for what scope, when, and (where traceable) from which raw events.

source_event_ids is populated on a best-effort basis (§27's own "What 'traceable' means in practice" note) -- an empty tuple is not an error; result, context_scope, and recorded_at alone already satisfy this package's accountability objective.

Examples:

>>> from datetime import timezone
>>> from mineproductivity.decision.result import DecisionResult
>>> entry = DecisionAuditEntry(
...     recorded_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
...     result=DecisionResult(model_code="STRATEGY.Threshold"),
...     context_scope={"pit": "north"}, source_event_ids=(),
... )
>>> dict(entry.context_scope)
{'pit': 'north'}

DecisionAuditTrail

DecisionAuditTrail()

Append-only record of every DecisionResult ever produced by any DecisionPipeline run in this process -- the accountability mechanism a business-decision system requires that a purely statistical one (analytics) does not as urgently. Serializes via core.serialization exactly as every other value object in this platform does.

Examples:

>>> from datetime import timezone
>>> from mineproductivity.decision.result import DecisionResult
>>> trail = DecisionAuditTrail()
>>> trail.record(DecisionAuditEntry(
...     recorded_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
...     result=DecisionResult(model_code="STRATEGY.Threshold"),
...     context_scope={"pit": "north"}, source_event_ids=(),
... ))
>>> len(trail.query())
1
>>> len(trail.query(scope={"pit": "south"}))
0

BatchDecisionRunner

BatchDecisionRunner(*, pipeline, audit_trail=None)

Runs one DecisionPipeline once over a bounded, already-assembled DecisionContext -- the 'normal,' scheduled- report mode, in contrast to RealTimeDecisionSession's live, event-driven mode (§25). A thin, named wrapper over DecisionPipeline.run (§9), exactly mirroring analytics.BatchAnalyticsRunner's relationship to AnalyticsPipeline (spec 06 §28).

Examples:

>>> from typing import ClassVar
>>> from mineproductivity.decision.abstractions import DecisionModel
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> from mineproductivity.decision.pipeline import ModelStage
>>> from mineproductivity.kpis import KPIResult
>>> class _Model(DecisionModel):
...     meta: ClassVar[DecisionMetadata] = DecisionMetadata(
...         code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
...     )
...     def _decide(self, context):
...         return DecisionResult(model_code="TEST")
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> runner = BatchDecisionRunner(pipeline=pipeline)
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
...                        analytics_results=(), scope={"pit": "north"})
>>> runner.run(ctx).unwrap().model_code
'TEST'

DecisionModelNotFoundError

DecisionModelNotFoundError(message, *, details=None)

Bases: NotFoundError

REGISTRY.get(code) found no registered DecisionModel for that code (design spec §32).

DecisionValidationError

DecisionValidationError(message, *, details=None)

Bases: ValidationError

A DecisionMetadata, Policy, or Threshold failed validation (design spec §30, §12, §13) -- e.g. an empty code, a Policy with zero rules, or an empty Threshold.field -- or a caller passed malformed arguments to a Decision method.

DecisionVersionConflictError

DecisionVersionConflictError(message, *, details=None)

Bases: RegistrationError

A plugin attempted to re-register an existing DecisionModel code with materially different metadata without a version bump (design spec §32), mirroring kpis.KPIVersionConflictError and analytics.AnalyticsVersionConflictError.

NoApplicablePolicyError

NoApplicablePolicyError(message, *, details=None)

Bases: NotFoundError

Raised only where a caller explicitly requests raising behavior instead of :meth:~mineproductivity.decision.abstractions.DecisionModel.decide's default empty-recommendation result for a context no policy applies to (design spec §8).

PolicyConflictError

PolicyConflictError(message, *, details=None)

Bases: RegistrationError

A plugin or governance action attempted to re-register an existing, Active Policy code with different rules/thresholds without a version bump and a Superseded transition for the prior version (design spec §12, §29) -- the Policy-layer analogue of :class:DecisionVersionConflictError, since a Policy is a separate governed artifact from a DecisionModel implementation (design spec §3.6).

ExplanationBuilder

Walks a Recommendation's triggered_rules/evidence and produces its Explanation. A concrete, non-pluggable utility -- not a DecisionModel subclass, since explanation construction is not a decision in itself.

Examples:

>>> from mineproductivity.kpis import KPIResult
>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(
...     kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
...     analytics_results=(), scope={},
... )
>>> explanation = ExplanationBuilder().build(rec, context=context)
>>> explanation.evidence_refs
('UTIL.OEE',)
>>> len(explanation.premises)
2

ExplanationStage

ExplanationStage(*, builder=None)

Bases: PipelineStage

The PipelineStage wrapper composing ExplanationBuilder into a DecisionPipeline -- attaches an Explanation to every Recommendation in context.recommendations, in place (see this module's own docstring for why this is non-terminal, matching §17's own typed signature).

Examples:

>>> from mineproductivity.kpis import KPIResult
>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(
...     kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
...     analytics_results=(), scope={}, recommendations=(rec,),
... )
>>> result = ExplanationStage().process(context)
>>> result.recommendations[0].explanation is not None
True

DecisionCategory

Bases: Enum

Closed enum -- adding a member is a governance-reviewed change, mirroring analytics.AnalyticsCategory's closed-enum rule (design spec §31).

DecisionMetadata dataclass

DecisionMetadata(code, *, name='', description, tags=frozenset(), attributes=dict(), category, version='1.0.0')

Bases: BaseMetadata

The minimal registration schema for a discoverable DecisionModel -- as light as analytics.AnalyticsMetadata, not as heavy as kpis.KPIMetadata, because a DecisionModel implementation is a computational strategy, not itself the governed business artifact (that weight belongs to Policy, design spec §12, §3.6).

BaseMetadata.name has no default upstream and a DecisionModel's code already serves as its identifier, so name defaults to code (via :meth:_normalize) whenever a caller does not supply one explicitly -- the same convention AnalyticsMetadata already established.

Examples:

>>> meta = DecisionMetadata(
...     code="STRATEGY.Threshold", category=DecisionCategory.STRATEGY,
...     description="Evaluate a Policy's rules against a DecisionContext.",
... )
>>> meta.code
'STRATEGY.Threshold'
>>> meta.name
'STRATEGY.Threshold'
>>> DecisionMetadata(code="", category=DecisionCategory.STRATEGY, description="x")
Traceback (most recent call last):
    ...
mineproductivity.decision.exceptions.DecisionValidationError: DecisionMetadata.code must not be empty

DecisionPipeline

DecisionPipeline(stages)

An ordered Sequence[PipelineStage], run in order over one input DecisionContext -- a typical pipeline gathers evidence, evaluates rules against an active Policy, generates recommendations, ranks them, and explains them.

A pipeline never contains strategy-specific branching (mirroring AnalyticsPipeline's "holds no model logic" invariant one layer down): :meth:run calls each stage's :meth:~PipelineStage.process uniformly and has no knowledge of which concrete DecisionModel a ModelStage wraps.

Examples:

>>> from typing import ClassVar
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> class _Model(DecisionModel):
...     meta: ClassVar[DecisionMetadata] = DecisionMetadata(
...         code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
...     )
...     def _decide(self, context):
...         return DecisionResult(model_code="TEST")
>>> from mineproductivity.kpis import KPIResult
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
...                        analytics_results=(), scope={})
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> pipeline.run(ctx).unwrap().model_code
'TEST'
>>> DecisionPipeline(stages=()).run(ctx).is_err
True

run

run(context)

Runs every stage in order; a non-terminal stage's output feeds the next stage's input. The last stage MUST be a ModelStage (or otherwise yield a DecisionResult) or this returns Result.err(DecisionValidationError(...)).

ModelStage

ModelStage(model)

Bases: PipelineStage

A terminal stage that hands the (by now rule-evaluated) context to one DecisionModel and yields its DecisionResult.

Examples:

>>> from typing import ClassVar
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> class _Model(DecisionModel):
...     meta: ClassVar[DecisionMetadata] = DecisionMetadata(
...         code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
...     )
...     def _decide(self, context):
...         return DecisionResult(model_code="TEST")
>>> stage = ModelStage(_Model())
>>> from mineproductivity.kpis import KPIResult
>>> ctx = DecisionContext(kpi_results=(KPIResult(code="PROD.TPH", value=1.0, unit="t/h"),),
...                        analytics_results=(), scope={})
>>> stage.process(ctx).model_code
'TEST'

PipelineStage

Bases: ABC

One step in a DecisionPipeline. Stateless and composable -- a new stage never requires changing DecisionPipeline itself.

Return type is deliberately DecisionContext | DecisionResult, not just DecisionContext: :class:ModelStage -- itself a PipelineStage -- always yields a DecisionResult, so the honest contract every implementation (terminal or not) must satisfy is the union of both. This mirrors analytics.pipeline.PipelineStage's identical, already-disclosed widening of its own design spec's abstract-method signature.

process abstractmethod

process(context)

Transform one DecisionContext into another (e.g. attaching rule-evaluation results) -- OR, for a terminal stage, wrap the context into a DecisionResult (see :class:ModelStage).

ActionPlanner

Sequences a set of prioritized actions, respecting declared dependencies, into an ActionPlan. Implements its own narrow topological ordering rather than reusing kpis.DependencyGraph, which is tightly coupled to Registry[str, type[BaseKPI]] and KPIMetadata.dependencies -- a shape that does not fit arbitrary action-to-action dependencies (§3.3, §21).

dependencies is an optional, caller-supplied mapping from an action's identifying key (Recommendation.policy_code) to the keys it depends on -- when empty (the default), :meth:plan degrades to simply ordering actions by ActionPriority.priority_score, so a caller with no dependency information to supply pays no complexity cost. A dependency entry naming a key absent from prioritized is silently ignored (the referenced action is not a candidate this call is planning over). :meth:plan returns Result.err on a detected cycle rather than silently dropping or arbitrarily breaking it -- the same "fail loudly on a cycle" posture kpis.DependencyGraph.detect_cycle already established for KPI dependencies, applied here to this package's own, structurally different dependency shape. For the identical reason, plan also returns Result.err when two entries in prioritized share a policy_code -- since the identifying key is the only handle dependencies (and this method's own de-duplication) has for an action, silently keeping only one would silently drop the other from the plan (a defect caught during this phase's own QA review, before it could ship).

Ordering uses a priority queue (heapq) over the ready frontier rather than a plain FIFO/DFS traversal, so the emitted order always respects priority_score (descending) among actions that are simultaneously eligible -- O((V+E) log V), a small, disclosed deviation from the plain O(V+E) bound a tie-break-free topological sort achieves, required because respecting priority order among ready actions is this method's own stated purpose, not an afterthought. Ties break by input order (a stable-sort-shaped guarantee, mirroring ranking.WeightedScoreRanking's own stable tie-break discipline).

Examples:

>>> from mineproductivity.decision.result import DecisionScore, Recommendation
>>> rec_a = Recommendation(
...     policy_code="A", triggered_rules=(), summary="Do A", severity="high", evidence=(),
... )
>>> rec_b = Recommendation(
...     policy_code="B", triggered_rules=(), summary="Do B", severity="medium", evidence=(),
... )
>>> ranked_a = RankedRecommendation(
...     recommendation=rec_a, score=DecisionScore(value=0.9, components={}), rank=1,
... )
>>> ranked_b = RankedRecommendation(
...     recommendation=rec_b, score=DecisionScore(value=0.5, components={}), rank=2,
... )
>>> priority_a = ActionPriority(urgency=0.75, impact=0.9, effort=1.0)
>>> priority_b = ActionPriority(urgency=0.5, impact=0.5, effort=1.0)
>>> plan = ActionPlanner().plan(
...     [(ranked_a, priority_a), (ranked_b, priority_b)], dependencies={"A": ("B",)},
... )
>>> [action.policy_code for action in plan.unwrap().ordered_actions]
['B', 'A']

DecisionStatus

Bases: Enum

The Policy lifecycle -- mirrors kpis.KPIStatus exactly, applied here to governed business artifacts rather than to individual computed results.

Policy dataclass

Policy(code, *, version='1.0.0', status=(lambda: DecisionStatus.PROPOSED)(), rules, thresholds=dict(), strategy_code)

Bases: BaseValueObject

A named, versioned bundle of :data:~mineproductivity.decision.rules.Rule\ s, the :class:~mineproductivity.decision.thresholds.Threshold\ s they reference, and which DecisionStrategy/category they activate -- the business-policy equivalent of kpis.KPIMetadata's governance weight.

Examples:

>>> from mineproductivity.core import PredicateSpecification
>>> policy = Policy(
...     code="AVAIL.LowFleetAvailability",
...     rules={"low_oee": PredicateSpecification(lambda ctx: True)},
...     strategy_code="STRATEGY.Threshold",
... )
>>> policy.status
<DecisionStatus.PROPOSED: 'proposed'>
>>> policy.version
'1.0.0'

ActionPrioritizer

ActionPrioritizer(*, effort_estimates=None, default_effort=1.0)

Given a Sequence[RankedRecommendation], assigns each an ActionPriority reflecting urgency/impact/effort under limited execution capacity.

Examples:

>>> from mineproductivity.decision.result import DecisionScore, Recommendation
>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="x", severity="high", evidence=(),
... )
>>> ranked = RankedRecommendation(
...     recommendation=rec, score=DecisionScore(value=0.7, components={}), rank=1,
... )
>>> prioritized = ActionPrioritizer().prioritize([ranked])
>>> item, priority = prioritized[0]
>>> priority.urgency, priority.impact
(0.75, 0.7)

RankingStrategy

Bases: DecisionModel, ABC

Category base for ranking strategies -- orders candidate Recommendation\ s by :class:~mineproductivity.decision.result.DecisionScore. Distinct from :class:~mineproductivity.decision.prioritization.ActionPrioritizer (§20): ranking answers "which recommendation is best," prioritization answers "given limited capacity, which action happens first" -- conflating the two is a recorded anti-pattern (§33).

WeightedScoreRanking

WeightedScoreRanking(*, scorer=None)

Bases: RankingStrategy

The default, concrete ranking strategy: orders Recommendation\ s by DecisionScore.value, descending. Ties preserve input relative order (Python's sorted() is stable).

Registered into decision.REGISTRY at import time, mirroring how strategy.ThresholdDecisionStrategy self-registers.

Examples:

>>> low = Recommendation(
...     policy_code="A", triggered_rules=("r",), summary="x", severity="low", evidence=(),
... )
>>> critical = Recommendation(
...     policy_code="B", triggered_rules=("r",), summary="x", severity="critical", evidence=(),
... )
>>> ranked = WeightedScoreRanking().rank([low, critical])
>>> [item.recommendation.policy_code for item in ranked]
['B', 'A']
>>> [item.rank for item in ranked]
[1, 2]

rank

rank(recommendations)

Scores every Recommendation via DecisionScorer, then orders them by DecisionScore.value, descending, assigning each its 1-based rank position.

RealTimeDecisionSession

RealTimeDecisionSession(*, bus, pipeline, kpi_engine, analytics_runner, kpi_codes, window, audit_trail=None)

A long-lived session that subscribes to an events.EventBus and, on each new relevant envelope, refreshes the necessary kpis.KPIEngine/analytics.BatchAnalyticsRunner outputs and re-runs a DecisionPipeline over the refreshed DecisionContext -- the live-operations-center counterpart of BatchDecisionRunner (§26), mirroring analytics.StreamingAnalyticsSession's role one layer down (spec 06 §27). Composes KPIEngine/BatchAnalyticsRunner rather than recomputing anything itself (§3.2).

Thread safety. Mirrors analytics.StreamingAnalyticsSession: the Event Framework's own concurrency contract makes no guarantee about which thread a subscriber's handler runs on, so _latest is guarded by one threading.Lock shared across every scope -- independent scopes serialize on this single lock, an intentionally simple posture for a per-process session object, not a high-throughput shared cache. Because a handler's own refresh work (KPIEngine.execute, BatchAnalyticsRunner.run, DecisionPipeline.run) happens outside the lock, two concurrent deliveries for the same scope could otherwise finish out of order and let an older envelope's slower handler overwrite a newer one's already-stored result -- caught during this phase's own QA review. _latest therefore stores each scope's envelope event_time_utc alongside its DecisionResult and only overwrites when the new envelope's event_time_utc is not older than what is already stored, reusing event_time_utc as the already-canonical "calculation basis" ordering signal (EventEnvelope's own docstring, events spec 01) rather than inventing a second sequencing concept.

Examples:

kpi_engine/analytics_runner are duck-typed fakes below, mirroring analytics.AnalyticsContext's own doctest precedent (its _FakeStore stand-in for event_store: EventStore[Any]) -- this keeps the example focused on RealTimeDecisionSession's own refresh-and-decide wiring rather than a full KPIEngine dependency graph.

>>> from typing import ClassVar
>>> from datetime import datetime, timezone
>>> from mineproductivity.core import Result
>>> from mineproductivity.decision.abstractions import DecisionModel
>>> from mineproductivity.decision.metadata import DecisionCategory, DecisionMetadata
>>> from mineproductivity.decision.pipeline import ModelStage
>>> from mineproductivity.events.bus import _InMemoryEventBus
>>> from mineproductivity.events.canonical import CycleEvent
>>> from mineproductivity.events.envelope import EventEnvelope, EventMetadata
>>> from mineproductivity.events.identifier import EventID
>>> from mineproductivity.events.versioning import EventVersion
>>> from mineproductivity.kpis import KPIResult
>>> class _Model(DecisionModel):
...     meta: ClassVar[DecisionMetadata] = DecisionMetadata(
...         code="STRATEGY.Doctest", category=DecisionCategory.STRATEGY, description="x",
...     )
...     def _decide(self, context):
...         return DecisionResult(model_code="TEST")
>>> class _FakeKPIEngine:
...     def execute(self, code, *, window, scope):
...         return Result.ok(KPIResult(code=code, value=1.0, unit="t/h", scope=scope))
>>> class _FakeAnalyticsRunner:
...     def run(self, series):
...         return Result.err("no analytics configured for this doctest")
>>> pipeline = DecisionPipeline(stages=(ModelStage(_Model()),))
>>> bus = _InMemoryEventBus()
>>> session = RealTimeDecisionSession(
...     bus=bus, pipeline=pipeline, kpi_engine=_FakeKPIEngine(),
...     analytics_runner=_FakeAnalyticsRunner(), kpi_codes=("PROD.TPH",), window="shift",
... )
>>> subscription = session.start()
>>> now = datetime(2026, 1, 1, tzinfo=timezone.utc)
>>> envelope = EventEnvelope(
...     event_id=EventID.generate(), version=EventVersion(),
...     payload=CycleEvent(equipment_id="HT-1", shift_id="A", queue_min=1.0,
...                         spot_min=0.5, load_min=2.0, haul_min=8.0, dump_min=1.0,
...                         return_min=6.0, payload_t=220.0),
...     event_time_utc=now, processing_time_utc=now, ingestion_time_utc=now,
...     metadata=EventMetadata(name="cycle", source_system="test"),
... )
>>> bus.publish(envelope)
>>> session.latest({"equipment_id": "HT-1", "shift": "A"}).model_code
'TEST'
>>> subscription.cancel()

start

start()

Subscribes to bus; each published EventEnvelope refreshes evidence for its scope and re-runs pipeline.

latest

latest(scope)

The most recent DecisionResult produced for scope, served without re-running the pipeline.

ActionPlan dataclass

ActionPlan(ordered_actions, priorities, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())

Bases: DecisionResult

A topologically-ordered sequence of prioritized actions, respecting declared dependencies (design spec §21).

Examples:

>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="Investigate fleet availability", severity="high", evidence=("UTIL.OEE",),
... )
>>> plan = ActionPlan(
...     ordered_actions=(rec,),
...     priorities={"AVAIL.LowFleetAvailability": ActionPriority(urgency=0.8, impact=0.6, effort=0.2)},
... )
>>> len(plan.ordered_actions)
1

ActionPriority dataclass

ActionPriority(urgency, impact, effort)

Bases: BaseValueObject

Urgency/impact/effort under limited execution capacity (design spec §20) -- a distinct question from ranking (§16). All three components are retained alongside the aggregate priority_score so an operator can audit the arithmetic, not merely trust it.

Examples:

>>> ActionPriority(urgency=0.8, impact=0.6, effort=0.2).priority_score
2.4

Alert dataclass

Alert(message, severity, scope, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), triggered_by=None)

Bases: DecisionResult

Produced from a ThresholdBreach or a high-severity Recommendation (design spec §22). AlertGenerator (a later phase, out of scope here) produces this value object only -- channel delivery is explicitly out of this package's scope.

Examples:

>>> Alert(message="Fleet availability critical", severity="critical",
...       scope={"pit": "north"}).severity
'critical'

ConfidenceScore dataclass

ConfidenceScore(value, basis)

Bases: BaseValueObject

The trust weight backing how strongly a recommendation should be acted on (design spec §24). basis records which evidence path produced value -- "data_quality"/"rule_strength"/ "combined" -- so a consumer can distinguish "we checked data quality and it was good" from "we had no data-quality signal to check."

Examples:

>>> ConfidenceScore(value=0.9, basis="data_quality").basis
'data_quality'

DecisionResult dataclass

DecisionResult(*, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())

Bases: BaseValueObject

The shared envelope every concrete Decision output composes -- mirrors analytics.AnalyticsResult's role, one layer up.

Examples:

>>> DecisionResult(model_code="STRATEGY.Threshold").model_code
'STRATEGY.Threshold'
>>> DecisionResult(warnings=("no evidence in context",)).warnings
('no evidence in context',)

DecisionScore dataclass

DecisionScore(value, components)

Bases: BaseValueObject

The numeric weight backing ranking (design spec §16, §23). components names the contributing factors, for auditability -- an operator reviewing a ranked list can see why one recommendation outranked another, not merely trust the aggregate.

Examples:

>>> DecisionScore(value=0.8, components={"severity": 0.5, "policy_weight": 0.3}).value
0.8

Explanation dataclass

Explanation(premises, evidence_refs)

Bases: BaseValueObject

A structured, evidence-linked justification -- never opaque prose only (design spec §17). premises are human-readable statements; evidence_refs are the exact KPIResult.code/ AnalyticsResult.model_code values a consuming system can re-fetch to verify the claim independently.

Deliberately a plain BaseValueObject, not a DecisionResult subclass -- it is a supporting value attached to a decision output, not a decision output itself (§28).

Examples:

>>> Explanation(premises=("OEE below 0.65",), evidence_refs=("UTIL.OEE",)).premises
('OEE below 0.65',)

RankedRecommendation dataclass

RankedRecommendation(recommendation, score, rank, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())

Bases: DecisionResult

One Recommendation ordered into a ranked sequence by DecisionScore (design spec §16).

Examples:

>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="Investigate fleet availability", severity="high", evidence=("UTIL.OEE",),
... )
>>> RankedRecommendation(
...     recommendation=rec, score=DecisionScore(value=0.8, components={}), rank=1,
... ).rank
1

Recommendation dataclass

Recommendation(policy_code, triggered_rules, summary, severity, evidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=(), explanation=None)

Bases: DecisionResult

One recommended action, produced by a DecisionStrategy (design spec §15) -- always traceable to the specific Policy/Rule names that produced it (policy_code, triggered_rules) and to the specific KPIResult/AnalyticsResult evidence it was computed from (evidence).

Examples:

>>> Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="Investigate fleet availability", severity="high",
...     evidence=("UTIL.OEE",),
... ).severity
'high'

RootCauseResult dataclass

RootCauseResult(candidate_causes, confidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())

Bases: DecisionResult

The outcome of a :class:~mineproductivity.decision.root_cause.RootCauseAnalyzer (design spec §18) -- a future phase's interface-only extension point; this result type is implemented now even though no producer of it ships in this phase.

Examples:

>>> RootCauseResult(
...     candidate_causes=("conveyor belt wear",),
...     confidence=ConfidenceScore(value=0.6, basis="rule_strength"),
... ).candidate_causes
('conveyor belt wear',)

ThresholdBreach dataclass

ThresholdBreach(threshold, observed_value, breached_at)

Bases: BaseValueObject

Records exactly which Threshold, at what observed value, was crossed (design spec §13, §28) -- the audit-trail-facing companion to Threshold itself.

Examples:

>>> from datetime import datetime, timezone
>>> breach = ThresholdBreach(
...     threshold=Threshold(field="value", comparator="<", limit=0.65),
...     observed_value=0.58,
...     breached_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
... )
>>> breach.observed_value
0.58

WhatIfResult dataclass

WhatIfResult(hypothesis, predicted_outcome, confidence, *, model_code='', computed_at=(lambda: datetime.now(timezone.utc))(), warnings=())

Bases: DecisionResult

The outcome of a :class:~mineproductivity.decision.what_if.WhatIfEngine (design spec §19) -- a future phase's interface-only extension point; this result type is implemented now even though no producer of it ships in this phase.

Examples:

>>> WhatIfResult(
...     hypothesis={"shift_length_hours": 10}, predicted_outcome="OEE improves by 2%",
...     confidence=ConfidenceScore(value=0.4, basis="rule_strength"),
... ).predicted_outcome
'OEE improves by 2%'

RootCauseAnalyzer

Bases: DecisionModel, ABC

The contract a future root-cause-analysis plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS -- identifying why a threshold was breached is a causal-inference problem this package's charter (§3.4) excludes.

Leaves :meth:~mineproductivity.decision.abstractions.DecisionModel._decide abstract (inherited, unoverridden) exactly as every other category base in this package does (DecisionStrategy, RankingStrategy) -- only a concrete subclass decides how its own _decide relates to _analyze, since no concrete subclass ships here to make that decision on a future plugin's behalf.

RuleEngine

Evaluates a named mapping of :data:Rule\ s against one DecisionContext, isolating one rule's evaluation failure from the rest.

Examples:

>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.kpis import KPIResult
>>> low_oee: Rule = PredicateSpecification(
...     lambda ctx: any(r.value is not None and r.value < 0.65 for r in ctx.kpi_results)
... )
>>> context = DecisionContext(
...     kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
...     analytics_results=(), scope={},
... )
>>> RuleEngine().evaluate({"low_oee": low_oee}, context)
('low_oee',)

evaluate

evaluate(rules, context)

Returns the names of every rule satisfied by context, in a stable (sorted) order. A single rule raising during is_satisfied_by is caught, logged, and excluded from the result -- never allowed to prevent evaluation of the remaining rules.

RuleEngineStage

RuleEngineStage(*, policy)

Bases: PipelineStage

The PipelineStage wrapper composing RuleEngine into a DecisionPipeline -- attaches the triggered rule names to the DecisionContext passed downstream, so ModelStage(ThresholdDecisionStrategy(policy=policy)) never has to re-run rule evaluation itself.

Examples:

>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.decision.policy import Policy
>>> policy = Policy(
...     code="TEST.Policy", rules={"always": PredicateSpecification(lambda ctx: True)},
...     strategy_code="STRATEGY.Threshold",
... )
>>> stage = RuleEngineStage(policy=policy)
>>> context = DecisionContext(kpi_results=(), analytics_results=(), scope={})
>>> stage.process(context).triggered_rules
('always',)

ConfidenceScorer

Computes :class:ConfidenceScore, deriving from analytics.DataQualityScore (spec 06 §25) plus how many/which rules corroborated a Recommendation -- never recomputing data quality itself. Falls back to "rule_strength" basis when no DataQualityScore is present in context.analytics_results.

This scorer's own derivation never produces a bare "data_quality" basis: a Recommendation always carries at least one triggered rule by construction (Policy.validate() requires at least one rule), so rule-strength evidence is always present alongside any DataQualityScore that may also be found -- the result is always "rule_strength" alone or "combined". The third basis literal ("data_quality") remains a legitimate value other producers may use; this class simply never has a reason to.

Examples:

>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="x", severity="high", evidence=("UTIL.OEE",),
... )
>>> context = DecisionContext(kpi_results=(), analytics_results=(), scope={})
>>> ConfidenceScorer().score(rec, context=context).basis
'rule_strength'

DecisionScorer

Computes the numeric weight (:class:DecisionScore) backing ranking (§16) -- a function of severity, policy weight (approximated here as rule-corroboration strength, see :func:_rule_strength), and, optionally, :class:ConfidenceScore (§24) -- never of raw KPI/Analytics values directly re-interpreted.

Examples:

>>> from mineproductivity.decision.result import ConfidenceScore
>>> rec = Recommendation(
...     policy_code="AVAIL.LowFleetAvailability", triggered_rules=("low_oee",),
...     summary="x", severity="critical", evidence=("UTIL.OEE",),
... )
>>> score = DecisionScorer().score(rec, confidence=ConfidenceScore(value=1.0, basis="rule_strength"))
>>> round(score.value, 4)
0.84

DecisionStrategy

Bases: DecisionModel, ABC

Category base for decision-making strategies -- the pluggable 'how do we decide' abstraction, analogous to analytics.TrendModel/BenchmarkModel one layer down.

ThresholdDecisionStrategy

ThresholdDecisionStrategy(*, policy, rule_engine=None, severity='medium')

Bases: DecisionStrategy

The default, concrete strategy: evaluate the active Policy's Rule\ s (via :class:~mineproductivity.decision.rules.RuleEngine) against the DecisionContext, and produce one Recommendation naming every triggered rule.

Registered into decision.REGISTRY at import time, mirroring how analytics.trend/analytics.baseline/analytics.benchmarking self-register.

severity is a constructor parameter, not a Policy/Threshold field: design spec §12's own Policy pseudocode (code, version, status, rules, thresholds, strategy_code) and §13's Threshold pseudocode (field, comparator, limit) carry no severity-bearing field anywhere, so there is nothing to read one from. A constructor parameter models "how bad is it when this policy's rules fire" as a property of the policy/strategy pairing (a reviewed, deliberate choice at configuration time), without adding an undocumented field to either already-specified, locked value object -- the smallest compatible resolution, disclosed here per this phase's own reconciliation rule.

Examples:

>>> from mineproductivity.core import PredicateSpecification
>>> from mineproductivity.decision.result import Recommendation
>>> from mineproductivity.decision.thresholds import Threshold
>>> from mineproductivity.kpis import KPIResult
>>> policy = Policy(
...     code="AVAIL.LowFleetAvailability",
...     rules={"low_oee": PredicateSpecification(
...         lambda ctx: any(r.value is not None and r.value < 0.65 for r in ctx.kpi_results)
...     )},
...     thresholds={"low_oee": Threshold(field="value", comparator="<", limit=0.65)},
...     strategy_code="STRATEGY.Threshold",
... )
>>> strategy = ThresholdDecisionStrategy(policy=policy, severity="high")
>>> context = DecisionContext(
...     kpi_results=(KPIResult(code="UTIL.OEE", value=0.58, unit=""),),
...     analytics_results=(), scope={"pit": "north"},
... )
>>> result = strategy.decide(context)
>>> isinstance(result, Recommendation)
True
>>> result.policy_code, result.triggered_rules, result.severity
('AVAIL.LowFleetAvailability', ('low_oee',), 'high')
>>> strategy.check_thresholds(context)[0].observed_value
0.58

check_thresholds

check_thresholds(context)

Independently checks every Threshold in self._policy.thresholds against context evidence, returning a ThresholdBreach for each one whose comparator/limit the resolved observed value satisfies. A Threshold whose field cannot be resolved against context (see :func:_resolve_field) is silently skipped, never raised -- "qualify, don't coerce".

Threshold dataclass

Threshold(field, comparator, limit)

Bases: BaseValueObject

One declarative limit a Rule references, e.g. Threshold(field="value", comparator="<", limit=0.65) checked against the relevant KPIResult/AnalyticsResult field named by field.

Deliberately data, not code -- a Policy can be updated to reference a new limit value through ordinary version governance (design spec §12) without touching any Rule's predicate logic.

Examples:

>>> Threshold(field="value", comparator="<", limit=0.65).limit
0.65
>>> Threshold(field="", comparator="<", limit=0.65)
Traceback (most recent call last):
    ...
mineproductivity.decision.exceptions.DecisionValidationError: Threshold.field must not be empty

WhatIfEngine

Bases: DecisionModel, ABC

The contract a future what-if/scenario-simulation plugin implements. THIS MODULE SHIPS NO CONCRETE SUBCLASS -- predicting the outcome of a hypothetical change is a simulation/forecasting problem this package's charter (§3.4) excludes.

Leaves :meth:~mineproductivity.decision.abstractions.DecisionModel._decide abstract (inherited, unoverridden) exactly as RootCauseAnalyzer does -- only a concrete subclass decides how its own _decide relates to _simulate, since no concrete subclass ships here to make that decision on a future plugin's behalf.

register

register(cls)

Register cls into :data:REGISTRY, keyed by cls.meta.code.

Raises:

Type Description
DecisionValidationError

If cls.meta.code is empty. In practice DecisionMetadata.validate() already rejects an empty code at construction time (§30), so this is a defensive, redundant check -- kept anyway for the same reason analytics.register/kpis.register keep their own equivalent check.

DecisionVersionConflictError

If cls.meta.code is already registered. Registry.register() is add-only and rejects any re-registration under an existing key, identical item or not (design spec AD-RG-04) -- this function does not attempt to distinguish "identical metadata, harmless re-import" from "materially different, dangerous repointing," exactly mirroring analytics.register's own behavior.