primus.optimization.portfolio_run module

Orchestrate one durable, human-governed optimization portfolio run.

This module intentionally contains no ranking, diagnosis, dispatch, or review algorithm. It composes the public optimization.* operations around one living Report and translates a structured Human.review response into the exact approved targets accepted by optimization.run.

The Tactus procedure is the transport for human_review. Keeping the orchestration transport-neutral makes checkpoint/replay deterministic and prevents a second, slightly different portfolio decision path from emerging.

class primus.optimization.portfolio_run.OptimizationPortfolioRunner(dependencies: PortfolioRunDependencies)

Bases: object

Run the safe, report-first portfolio lifecycle once or through replay.

human_review may suspend through Tactus. On replay, the Tactus checkpoint supplies the same accepted structured response, while the report service resumes the same deterministic run key and report URL.

__init__(dependencies: PortfolioRunDependencies) → None

run(request: Mapping[str, Any]) → dict[str, Any]

class primus.optimization.portfolio_run.PortfolioRunDependencies(rank: Callable[[dict[str, Any]], Mapping[str, Any]], assess: Callable[[dict[str, Any]], Mapping[str, Any]], diagnose: Callable[[dict[str, Any]], Mapping[str, Any]], summary: Callable[[dict[str, Any]], Mapping[str, Any]], dispatch: Callable[[dict[str, Any]], Mapping[str, Any]], review: Callable[[dict[str, Any]], Mapping[str, Any]], report_service: Callable[[str, Mapping[str, Any]], Any], human_review: Callable[[dict[str, Any]], Mapping[str, Any]], create_action: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None, publish_update: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None, supersede_action: Callable[[str, str], Mapping[str, Any]] | None = None, optimizer_child_step: Callable[[...], Mapping[str, Any]] | None = None, optimizer_child_request: Callable[[Mapping[str, Any]], Mapping[str, Any]] | None = None, diagnosis_preflight: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None)

Bases: object

Transport adapters for the existing public optimization capabilities.

__init__(rank: Callable[[dict[str, Any]], Mapping[str, Any]], assess: Callable[[dict[str, Any]], Mapping[str, Any]], diagnose: Callable[[dict[str, Any]], Mapping[str, Any]], summary: Callable[[dict[str, Any]], Mapping[str, Any]], dispatch: Callable[[dict[str, Any]], Mapping[str, Any]], review: Callable[[dict[str, Any]], Mapping[str, Any]], report_service: Callable[[str, Mapping[str, Any]], Any], human_review: Callable[[dict[str, Any]], Mapping[str, Any]], create_action: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None, publish_update: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None, supersede_action: Callable[[str, str], Mapping[str, Any]] | None = None, optimizer_child_step: Callable[[...], Mapping[str, Any]] | None = None, optimizer_child_request: Callable[[Mapping[str, Any]], Mapping[str, Any]] | None = None, diagnosis_preflight: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None) → None

assess: Callable[[dict[str, Any]], Mapping[str, Any]]

create_action: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None

diagnose: Callable[[dict[str, Any]], Mapping[str, Any]]

diagnosis_preflight: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None

dispatch: Callable[[dict[str, Any]], Mapping[str, Any]]

human_review: Callable[[dict[str, Any]], Mapping[str, Any]]

optimizer_child_request: Callable[[Mapping[str, Any]], Mapping[str, Any]] | None = None

optimizer_child_step: Callable[[...], Mapping[str, Any]] | None = None

publish_update: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None

rank: Callable[[dict[str, Any]], Mapping[str, Any]]

report_service: Callable[[str, Mapping[str, Any]], Any]

review: Callable[[dict[str, Any]], Mapping[str, Any]]

summary: Callable[[dict[str, Any]], Mapping[str, Any]]

supersede_action: Callable[[str, str], Mapping[str, Any]] | None = None

primus.optimization.portfolio_run.drive_optimizer_child_launch(request: Mapping[str, Any], *, initial_state: Mapping[str, Any] | None, step: Callable[[...], Mapping[str, Any]], publish: Callable[[Mapping[str, Any]], Any]) → dict[str, Any]

Publish each child state before allowing its following mutation.

step is deliberately Report-agnostic. may_mutate is true only when this process successfully published the immediately preceding state; a replay starts read/reconcile-only until it reaches a newly published boundary.