primus.optimization.run_report module

Durable, append-only reporting for one periodic optimization run.

The service deliberately owns no optimization orchestration. Callers execute the decision stages with persist=False and publish their returned packets at well-defined milestones. A single Task is the lifecycle authority; one Report is its stable stakeholder location; ReportBlocks contain only compact pointers to immutable JSON/XLSX revisions.

exception primus.optimization.run_report.OptimizationRunIntegrityError

Bases: OptimizationRunPublicationError

Committed recovery evidence is corrupt or does not identify this run.

exception primus.optimization.run_report.OptimizationRunPublicationError

Bases: RuntimeError

A milestone could not be made durable, so the run must stop.

class primus.optimization.run_report.OptimizationRunReportService(*, client: ~typing.Any, account_id: str, run_key: str, report_configuration_id: str | None = None, dashboard_base_url: str | None = None, existing_task: ~typing.Any = None, now: ~typing.Callable[[], ~datetime.datetime] = <function _utc_now>, task_lookup: ~typing.Callable[[str], ~typing.Any] | None = None, report_lookup: ~typing.Callable[[~typing.Any], ~typing.Any] | None = None, block_lookup: ~typing.Callable[[~typing.Any], list[~typing.Any]] | None = None, stage_lookup: ~typing.Callable[[~typing.Any], list[~typing.Any]] | None = None, artifact_uploader: ~typing.Callable[[str, str, bytes], str] | None = None, artifact_store: ~typing.Any | None = None, verify_uploaded_artifacts: bool = True, attempt_id_factory: ~typing.Callable[[], str] = <function OptimizationRunReportService.<lambda>>, publication_id_factory: ~typing.Callable[[], str] = <function OptimizationRunReportService.<lambda>>)

Bases: object

Publish a periodic run to one stable Report and append-only revisions.

__init__(*, client: ~typing.Any, account_id: str, run_key: str, report_configuration_id: str | None = None, dashboard_base_url: str | None = None, existing_task: ~typing.Any = None, now: ~typing.Callable[[], ~datetime.datetime] = <function _utc_now>, task_lookup: ~typing.Callable[[str], ~typing.Any] | None = None, report_lookup: ~typing.Callable[[~typing.Any], ~typing.Any] | None = None, block_lookup: ~typing.Callable[[~typing.Any], list[~typing.Any]] | None = None, stage_lookup: ~typing.Callable[[~typing.Any], list[~typing.Any]] | None = None, artifact_uploader: ~typing.Callable[[str, str, bytes], str] | None = None, artifact_store: ~typing.Any | None = None, verify_uploaded_artifacts: bool = True, attempt_id_factory: ~typing.Callable[[], str] = <function OptimizationRunReportService.<lambda>>, publication_id_factory: ~typing.Callable[[], str] = <function OptimizationRunReportService.<lambda>>) → None

fail(message: str) → OptimizationRunReportState

finalize(*, status: str = 'completed') → OptimizationRunReportState

Finish a run while preserving its stakeholder-visible terminal state.

Task.status has no distinct incomplete/blocked terminal values, so it remains COMPLETED for a safely concluded non-failure run. The precise lifecycle outcome is retained in immutable task metadata and the Report cover page rather than being mislabeled as a failure.

load_latest_checkpoint() → dict[str, Any] | None

Load and verify the last committed evidence revision for replay.

The Report parameter update performed by _record_latest_revision is the publication commit point. Unreferenced uploads from an interrupted publication are deliberately ignored. Reads use the same GraphQL-authorized Task attachment route as every other artifact.

load_semantic_budget_ledger() → dict[str, Any] | None: Load and checksum-verify the Report’s committed semantic ledger.

persist_semantic_budget_ledger(ledger_value: Mapping[str, Any]) → dict[str, Any]

Commit one immutable semantic-ledger revision without a workbook.

Uploading and attaching the JSON may leave an orphan when publication is interrupted. The pointer under Report.parameters is the sole commit point, so recovery deliberately ignores every unreferenced upload.

publish_milestone(milestone: str, decision_evidence: Mapping[str, Any], *, stakeholder_view: Mapping[str, Any]) → PublishedRevision

publish_progress(*, phase: str, subphase: str | None = None, current: int, total: int | None, message: str, unit: str | None = None, state: str = 'active', elapsed_seconds: int | None = None, next_checkpoint: str | None = None, heartbeat_interval_seconds: int | None = None, artifact_counts: Mapping[str, Mapping[str, int]] | None = None, failure: Mapping[str, Any] | None = None) → None

Publish lightweight live progress without creating an evidence revision.

Milestones remain the immutable audit trail. This method updates only the existing analysis stage, compact status block, and Report cover so operators can see movement during long assessment and diagnosis loops.

start_or_resume(run_spec: Mapping[str, Any]) → OptimizationRunReportState

class primus.optimization.run_report.OptimizationRunReportState(task: 'Any', report: 'Any', blocks: 'dict[str, Any]', stages: 'dict[str, Any]', run_spec: 'Mapping[str, Any]', attempt_id: 'str', operator_identity: 'OptimizationOperatorIdentity')

Bases: object

__init__(task: Any, report: Any, blocks: dict[str, Any], stages: dict[str, Any], run_spec: Mapping[str, Any], attempt_id: str, operator_identity: OptimizationOperatorIdentity) → None

attempt_id: str

blocks: dict[str, Any]

operator_identity: OptimizationOperatorIdentity

report: Any

run_spec: Mapping[str, Any]

stages: dict[str, Any]

task: Any

exception primus.optimization.run_report.OptimizationRunRetryablePublicationError

Bases: OptimizationRunPublicationError

A credential or publication interruption left no new committed revision.

Callers must keep the Task active and replay from latest_revision.

class primus.optimization.run_report.PublishedRevision(number: 'int', milestone: 'str', published_at: 'str', raw_evidence_path: 'str', workbook_path: 'str', manifest_path: 'str', manifest_checksum: 'str', manifest_size_bytes: 'int', evidence_checksum: 'str', workbook_checksum: 'str', row_counts: 'Mapping[str, int]', overview: 'Mapping[str, Any]', artifacts: 'tuple[Mapping[str, Any], ...]' = (), detail_status: 'str' = 'pending', detail_source_revision: 'Optional[int]' = None)

Bases: object

__init__(number: int, milestone: str, published_at: str, raw_evidence_path: str, workbook_path: str, manifest_path: str, manifest_checksum: str, manifest_size_bytes: int, evidence_checksum: str, workbook_checksum: str, row_counts: Mapping[str, int], overview: Mapping[str, Any], artifacts: tuple[Mapping[str, Any], ...] = (), detail_status: str = 'pending', detail_source_revision: int | None = None) → None

artifacts: tuple[Mapping[str, Any], ...] = ()

detail_source_revision: int | None = None

detail_status: str = 'pending'

evidence_checksum: str

manifest_checksum: str

manifest_path: str

manifest_size_bytes: int

milestone: str

number: int

overview: Mapping[str, Any]

published_at: str

raw_evidence_path: str

row_counts: Mapping[str, int]

workbook_checksum: str

workbook_path: str

class primus.optimization.run_report.WorkbookArtifact(content: 'bytes', checksum: 'str', row_counts: 'Mapping[str, int]')

Bases: object

__init__(content: bytes, checksum: str, row_counts: Mapping[str, int]) → None

checksum: str

content: bytes

row_counts: Mapping[str, int]

primus.optimization.run_report.build_scorecard_artifacts(stakeholder_view: Mapping[str, Any], *, revision_number: int, task_id: str, uploader: Callable[[str, str, bytes], str], publication_id: str | None = None, on_artifact_upload: Callable[[str, bool], None] | None = None, reuse_artifact: Callable[[str, str, str, bytes, str], Mapping[str, Any] | None] | None = None, on_artifact_resolved: Callable[[str, Mapping[str, Any]], None] | None = None, agent_followups: Sequence[Mapping[str, Any]] = ()) → list[dict[str, Any]]: Publish one safe Markdown summary and quantitative CSV per scorecard.

primus.optimization.run_report.build_stakeholder_presentation(stakeholder_view: Mapping[str, Any], *, scorecard_artifacts: list[Mapping[str, Any]], detail_status: str = 'not_requested', detail_source_revision: int | None = None) → dict[str, Any]: Build the safe, deterministic aggregate/card projection for the dashboard.

primus.optimization.run_report.build_stakeholder_workbook(stakeholder_view: Mapping[str, Any], *, revision_number: int, generated_at: datetime) → WorkbookArtifact: Create the fixed, macro-free stakeholder workbook from a safe projection only.

primus.optimization.run_report.dashboard_base_url_from_account_settings(settings: Any) → str | None

primus.optimization.run_report.optimization_run_key(account_id: str, run_spec: Mapping[str, Any]) → str: Stable identity for a frozen account/scope/policy run.