primus.optimization package

Deterministic, transport-independent optimization decision helpers.

class primus.optimization.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.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.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.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.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.normalize_structural_state(raw: Any) → str: Normalize champion/configuration/terminal-class checks into one state.

primus.optimization.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.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.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.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.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 package

Submodules