primus.optimization.decision module

Pure decision logic for the optimization toolchain.

This module deliberately accepts and returns plain dictionaries and lists. It contains no API access, persistence, model calls, or score mutations so the MCP and CLI surfaces can share identical, reproducible decisions.

class primus.optimization.decision.OptimizationDecisionPacket(account_id: str | None, scope: ~typing.Mapping[str, ~typing.Any], window: ~typing.Mapping[str, ~typing.Any], evidence: ~typing.Mapping[str, ~typing.Any], states: ~typing.Mapping[str, str], primary_next_action: str, policy_version: str = 'feedback-investment-v1', champion_version: str | None = None, feedback_watermark: str | None = None, secondary_actions: ~typing.Sequence[str] = <factory>, blockers: ~typing.Sequence[str] = <factory>, evidence_ids: ~typing.Sequence[str] = <factory>, rationale: str = '', evidence_fingerprint: str | None = None, stakeholder_questions: ~typing.Sequence[str] = <factory>)

Bases: object

Versioned portable result envelope shared by every toolchain stage.

__init__(account_id: str | None, scope: ~typing.Mapping[str, ~typing.Any], window: ~typing.Mapping[str, ~typing.Any], evidence: ~typing.Mapping[str, ~typing.Any], states: ~typing.Mapping[str, str], primary_next_action: str, policy_version: str = 'feedback-investment-v1', champion_version: str | None = None, feedback_watermark: str | None = None, secondary_actions: ~typing.Sequence[str] = <factory>, blockers: ~typing.Sequence[str] = <factory>, evidence_ids: ~typing.Sequence[str] = <factory>, rationale: str = '', evidence_fingerprint: str | None = None, stakeholder_questions: ~typing.Sequence[str] = <factory>) → None

account_id: str | None

blockers: Sequence[str]

champion_version: str | None = None

evidence: Mapping[str, Any]

evidence_fingerprint: str | None = None

evidence_ids: Sequence[str]

feedback_watermark: str | None = None

policy_version: str = 'feedback-investment-v1'

primary_next_action: str

rationale: str = ''

scope: Mapping[str, Any]

secondary_actions: Sequence[str]

stakeholder_questions: Sequence[str]

states: Mapping[str, str]

to_dict() → dict[str, Any]

window: Mapping[str, Any]

primus.optimization.decision.assess_investment(evidence: Mapping[str, Any], *, policy: Mapping[str, Any] | None = None, context: Mapping[str, Any] | None = None) → dict[str, Any]

Apply feedback-investment-v1 gates in documented safety-first order.

Ordering is: coverage, mechanical structure, guideline/semantic defects, stakeholder questions, feedback curation, count, class reachability, stability, then Wilson-backed disagreement outcome.

primus.optimization.decision.bounded_diagnostic_assessment_failure(assessment: Mapping[str, Any], *, max_samples: int) → str | None: Fail closed unless a non-ready score is safe for a bounded experiment.

primus.optimization.decision.classify_post_run_review(evidence: Mapping[str, Any], *, context: Mapping[str, Any] | None = None) → dict[str, Any]: Classify optimizer evidence; it never promotes anything itself.

primus.optimization.decision.dispatch_optimization_operation(operation: str, payload: Mapping[str, Any], **dependencies: Any) → dict[str, Any]

Route transport payloads to pure operations without accessing services.

Runtime adapters own pagination, semantic jobs, persistence, and optimizer dispatch. This router intentionally handles only deterministic decisions.

primus.optimization.decision.evaluate_score_activity(score: Mapping[str, Any], *, as_of: datetime | str | None = None) → dict[str, Any]

Evaluate the fixed rolling cooldown from complete inventory evidence.

Both the mutable score timestamp and the newest immutable version timestamp are required. Missing or malformed evidence fails closed. Future timestamps are conservatively treated as recent rather than silently repairing clock skew.

primus.optimization.decision.evidence_fingerprint(evidence: Mapping[str, Any]) → str: Hash canonical evidence so a caller can reject stale assessments.

primus.optimization.decision.frozen_utc_window(now: datetime | None = None, complete_days: int | None = None) → dict[str, Any]: Return the preceding complete UTC days; the current partial day is excluded.

primus.optimization.decision.normalize_diagnosis(diagnosis: Mapping[str, Any] | None, *, context: Mapping[str, Any] | None = None) → dict[str, Any]: Normalize asynchronous semantic diagnosis without inventing a model decision.

primus.optimization.decision.normalize_execution_candidate_policy(value: Any) → str

Return the explicit, frozen execution-candidate policy.

The conservative policy is deliberately the default for every public caller. Diagnostic optimization is a separately named opt-in; it never changes promotion eligibility or the post-run promotion evidence gates.

primus.optimization.decision.normalize_guideline_state(raw: Any) → str: Normalize mechanical/semantic adapter results into the public state set.

primus.optimization.decision.normalize_rank_scope(request: Mapping[str, Any] | None) → dict[str, list[str]]

Validate and normalize optional scorecard selectors for portfolio ranking.

Omitted selectors mean account-wide ranking. Explicit selector arrays must each contain at least one non-blank string so a malformed scoped request can never widen silently to the whole account. Values are deduplicated without changing opaque IDs. Prefixes are case-folded because their matching semantics are case-insensitive and equivalent scopes need one fingerprint.

primus.optimization.decision.normalize_structural_state(raw: Any) → str: Normalize champion/configuration/terminal-class checks into one state.

primus.optimization.decision.rank_opportunities(scores: Sequence[Mapping[str, Any]], *, coverage: Mapping[str, Any] | None = None, context: Mapping[str, Any] | None = None) → dict[str, Any]: Rank fully analyzed scores without silently treating partial enumeration as exact.

primus.optimization.decision.rank_portfolio(scores: Sequence[Mapping[str, Any]], *, coverage: Mapping[str, Any] | None = None, context: Mapping[str, Any] | None = None) → dict[str, Any]: Rank fully analyzed scores without silently treating partial enumeration as exact.

primus.optimization.decision.rank_scope_matches(scorecard_id: Any, scorecard_name: Any, scope: Mapping[str, Sequence[str]] | None) → bool: Return whether one scorecard belongs to a normalized rank scope.

primus.optimization.decision.review_optimizer_result(evidence: Mapping[str, Any], *, context: Mapping[str, Any] | None = None) → dict[str, Any]: Classify optimizer evidence; it never promotes anything itself.

primus.optimization.decision.run_operation(operation: str, payload: Mapping[str, Any], **dependencies: Any) → dict[str, Any]

Route transport payloads to pure operations without accessing services.

Runtime adapters own pagination, semantic jobs, persistence, and optimizer dispatch. This router intentionally handles only deterministic decisions.

primus.optimization.decision.summarize_packets(packets: Sequence[Mapping[str, Any]], *, context: Mapping[str, Any] | None = None) → dict[str, Any]: Produce a compact, deterministic portfolio summary from decision packets.

primus.optimization.decision.validate_approved_batch(targets: Sequence[Mapping[str, Any]], *, approved: bool, current_fingerprints: Mapping[str, str] | None = None, execution_candidate_policy: str = 'promotion_ready', max_samples: int | None = None, context: Mapping[str, Any] | None = None) → dict[str, Any]: Validate an explicit <=5 target approval and reject stale targets individually.

primus.optimization.decision.validate_public_run_dispatch(payload: Mapping[str, Any]) → dict[str, Any]: Apply limits and assessment provenance before lower-level batch checks.

primus.optimization.decision.validate_run_limits(payload: Mapping[str, Any]) → dict[str, Any]

Validate public optimizer-dispatch limits once for every transport.

Cost may be a positive real number. Sample and iteration caps are positive integers, and concurrency is an explicitly bounded integer from one to five. Missing values are never defaulted.

primus.optimization.decision.weekly_buckets(timestamps: Iterable[str | datetime], *, window_end: str | datetime, weeks: int | None = None) → list[dict[str, Any]]: Bucket timestamps into complete Monday-based UTC weeks ending at window_end.

primus.optimization.decision.wilson_interval(successes: int | float, total: int | float, confidence: float = 0.95) → tuple[float, float]: Two-sided Wilson score interval, including exact zero and one endpoints.