Platform Services

Persistent workbench state shared by the agents and the API server: the sessions store (SQLite), run-metric history, and read-side services. Dependency-light by design (no web framework).

Platform services: persistent workbench state shared by the agents and the API server.

This subpackage is the dependency-light persistence/service layer that used to live in mmm_framework.api next to the FastAPI app. The web app itself moved to the separate mmm-framework-server package (server/ in the repo); what remains here is business state, not web routing:

• sessions — the sessions store (SQLite): orgs/users, projects, chat sessions, model runs, artifacts, the experiment lifecycle registry, preferences/branding.

• history — per-run metric persistence.

• runs, pacing, scorecard, triangulation, estimands, portfolio_benchmark — read-side services over that state.

• backfill, backup, connection_sync, onboarding, observability — maintenance / integration helpers.

Nothing in this subpackage imports FastAPI or the LLM stack at module level; it must stay importable in a lean pip install mmm-framework environment.

Note: platform shadows nothing for external code — absolute imports mean import platform anywhere still resolves to the stdlib module.

Sessions

Session metadata store + per-session artifact log.

The langgraph SqliteSaver owns its own tables for checkpoints. We add two sibling tables in the same DB file:

sessions(thread_id PK, name, created_at, updated_at) artifacts(id PK, thread_id FK, kind, payload_json, created_at)

kind is one of: ‘code_snippet’, ‘report’, ‘plot’, ‘saved_config’, ‘saved_model’. payload_json is whatever the frontend needs to render or rerun the artifact (e.g. {“code”: “…”} for code_snippet).

mmm_framework.platform.sessions.resolve_db_path()

Resolve the sessions DB location.

MMM_SESSIONS_DB overrides the package-local default so deployments can keep state on a persistent disk without symlinking into the install tree (see deploy/gcp/vm/vm_setup.sh). The server’s async checkpointer (mmm_framework_server.main) and auth/store.py honor the same variable — all three must point at the same file.

Before this module moved from mmm_framework.api to mmm_framework.platform (2026-07-24), the package-local default lived in the old api/ directory. An existing dev DB at that legacy path keeps winning so the move does not silently orphan local sessions/projects.

Return type:

Path

mmm_framework.platform.sessions.init_db()

Return type:

None

mmm_framework.platform.sessions.list_sessions(project_id=None)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_session(thread_id)

Return a single session row with artifact_count, or None if not found.

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.create_session(name=None, project_id=None, modeling_mode=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.update_session(thread_id, name=None, project_id=None, modeling_mode=None)

Update session name, project_id and/or modeling_mode. Returns True if found.

Return type:

bool

mmm_framework.platform.sessions.touch_session(thread_id)

Bump updated_at; if the row doesn’t exist (legacy thread), insert it.

Return type:

None

mmm_framework.platform.sessions.rename_session(thread_id, name)

Return type:

bool

mmm_framework.platform.sessions.delete_session(thread_id)

Return type:

bool

mmm_framework.platform.sessions.add_artifact(thread_id, kind, payload)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_artifacts(thread_id)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_artifact(artifact_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.delete_artifact(artifact_id)

Return type:

bool

mmm_framework.platform.sessions.update_artifact_payload(artifact_id, payload)

Overwrite an artifact’s payload (used by the async experiment-simulation job to flip status pending → running → done/error in place).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.record_assumption(thread_id, key, value, rationale, category='other', change_note=None)

Create or update an assumption. New version always appended; history preserved.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.retract_assumption(thread_id, key, reason)

Mark an assumption as retracted (tombstone); preserved in history.

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_assumptions(thread_id, include_history=False)

If include_history is False, returns the latest version per key. Otherwise returns every row ordered by (key, version).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_assumption_history(thread_id, key)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.set_workflow_step(thread_id, step, status, notes=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_workflow_overrides(thread_id)

Return type:

dict[int, dict[str, Any]]

mmm_framework.platform.sessions.register_file(thread_id, path, name, kind, size_bytes=None, preview=None, meta=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_files(thread_id)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_file(file_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.delete_file(file_id)

Return type:

bool

mmm_framework.platform.sessions.lock_analysis_plan(thread_id, name, payload)

Snapshot the current analysis plan (research question + DAG + assumptions) as a locked record.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_analysis_plans(thread_id=None)

List analysis plans for a thread (or all plans if thread_id is None).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_analysis_plan(plan_id)

Get a single analysis plan by ID.

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.delete_analysis_plan(plan_id)

Return type:

bool

mmm_framework.platform.sessions.upsert_experiment(*, experiment_id=None, project_id=None, thread_id=None, channel=None, subchannel=None, design_type=None, status=None, start_date=None, end_date=None, estimand=None, value=None, se=None, notes=None, recommending_run_id=None, calibrated_run_id=None, design=None, readout=None, priority=None, allow_calibrated_edit=False)

Create (no experiment_id) or partially update an experiment record.

On update, only the non-None fields change. Raises ValueError for an unknown id, a missing channel on create, or an invalid status.

Status is state-machine enforced: creating directly in ‘calibrated’ is rejected (calibration happens via transition_experiment() at fit close-out); creating in planned/running/completed stays legal (historical import) but the full draft→…→status history chain is backfilled. An update may change status to anything REACHABLE through ALLOWED_TRANSITIONS — multi-hop moves (e.g. planned→completed, the one-call results-recording flow) backfill the intermediate hops into the history — EXCEPT ‘calibrated’, which is only ever set by transition_experiment() (fit close-out). Same-status (or status=None) updates stay silent no-ops for the other fields.

A calibrated experiment’s measurement fields (value / se / estimand / start_date / end_date / readout / channel / subchannel) feed the model’s calibration likelihood, so CHANGING any of them raises unless allow_calibrated_edit=True (the record_experiment_readout tool’s sanctioned, audited overwrite path). Unchanged re-sends and notes/design/priority-only updates stay allowed.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_experiment(experiment_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.transition_experiment(experiment_id, new_status, *, note=None, value=None, se=None, estimand=None, start_date=None, end_date=None, readout=None, calibrated_run_id=None)

Validated lifecycle move with an append-only audit trail.

Enforces ALLOWED_TRANSITIONS (ValueError on an illegal move); stamps preregistered_at on draft→planned; merges the lift readout fields on →completed; records calibrated_run_id on →calibrated.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.append_experiment_event(experiment_id, note, changed=None)

Append a non-transition audit event to an experiment’s history.

Records an entry {status: <current status>, at, note[, changed]} in status_history_json without moving the state machine — used to leave a trail when a readout is edited in place (e.g. re-recording a value or attaching off-panel spend on an already-measured experiment). NULL-tolerant like transition_experiment() (pre-lifecycle rows get their history synthesized from the current status). Bumps updated_at.

Parameters:

• experiment_id (str) – Registry id of the experiment.

• note (str) – Human-readable description of the event.

• changed (dict[str, Any] | None) – Optional {field: [old, new]} diff of what changed.

Raises:

ValueError – For an unknown experiment id.

Return type:

None

mmm_framework.platform.sessions.latest_calibrated_evidence(project_id=None)

Newest calibrated experiment per channel — the evidence map that feeds information decay and the calibration-coverage view.

Returns {channel: {experiment_id, end_date, calibrated_run_id, se, estimand, updated_at}}. Recency is judged by end_date when present, falling back to updated_at.

Return type:

dict[str, dict]

mmm_framework.platform.sessions.list_experiments(project_id=None, status=None, channel=None, subchannel=None)

Experiments, newest-updated first; optionally filtered.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.delete_experiment(experiment_id)

Return type:

bool

mmm_framework.platform.sessions.create_learning_program(*, project_id=None, thread_id=None, name=None, channels, config, state_path=None, status='active')

Insert a learning-program row. channels is the FLATTENED arm list (the surface dimensions); config is the dollars-at-the-boundary program config (wiring contract §3.1).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_learning_program(program_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_learning_programs(project_id=None, status=None)

Learning programs, newest-updated first; optionally filtered.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.update_learning_program(program_id, *, name=None, status=None, thread_id=None, channels=None, config=None, state_path=None, summary=None)

Partial update; only the non-None fields change. Raises ValueError for an unknown id or an invalid status.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.delete_learning_program(program_id)

Delete a program and cascade its waves. Returns True if the program existed. (The on-disk state.npz is left for the caller to reap.)

Return type:

bool

mmm_framework.platform.sessions.add_learning_wave(program_id, *, project_id=None, wave_index=None, status='designed', source=None, design=None, observations=None, snapshot=None, experiment_ids=None)

Append a wave row (wave_index auto-increments per program).

source is one of wave / experiment_import / manual; snapshot is the pinned fit_and_plan SNAPSHOT (immutable, like run_metrics rows).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_learning_wave(wave_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_learning_waves(program_id)

A program’s waves, oldest first (wave_index ascending).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.update_learning_wave(wave_id, *, status=None, source=None, design=None, observations=None, snapshot=None, experiment_ids=None)

Partial update of a wave row (e.g. designed → ingested with results).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.record_ingested_wave(program_id, *, project_id=None, source=None, observations=None, snapshot=None, experiment_ids=None)

Record ingested evidence on the wave board (designed → ingested).

A rows-ingest (source='wave') RESOLVES the program’s latest wave row when that row is still 'designed' — so the timeline shows ONE row per real wave instead of a permanently-open ‘designed’ card plus a duplicate ‘ingested’ one. Experiment imports (and ingests with no open design) append a new 'ingested' row. Shared by the REST fit worker and the agent tools’ record_learning_wave/import_past_experiments paths.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.upsert_budget_plan(*, plan_id=None, project_id=None, org_id, name, description=None, model_id=None, kind='optimization', spend_changes=None, baseline_outcome=None, scenario_outcome=None, outcome_change=None, outcome_change_pct=None, channel_details=None, plan_payload=None)

Create or update a budget plan; returns the stored plan dict.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_budget_plan(plan_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_budget_plans(org_id, project_id=None, model_id=None)

Plans for an org (required), newest-updated first; optionally filtered.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.delete_budget_plan(plan_id)

Return type:

bool

mmm_framework.platform.sessions.latest_budget_plan_for_project(project_id)

The most recently updated saved budget plan for a project (project-scoped, org-agnostic) — the pacing loop auto-sources its planned series from this (issue #123). None when the project has no saved plan.

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.upsert_delivery(project_id, records, *, source=None)

Upsert actual-delivery rows for a project (issue #123).

Each record is {channel, spend, period?}; a row is keyed by (project_id, channel, period) so re-uploading a period overwrites it (the elapsed window can be re-stated as more actuals land). Records with a non-numeric or missing spend/channel are skipped. Returns all of the project’s delivery rows after the write.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.list_delivery(project_id)

A project’s actual-delivery rows, oldest-first within a channel so a period series reads in order (period is a free-form label — callers align by it).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.delete_delivery(project_id, *, channel=None, period=None)

Delete delivery rows for a project (optionally one channel / period). Returns the number of rows removed.

Return type:

int

mmm_framework.platform.sessions.upsert_platform_figures(project_id, records)

Upsert platform-reported attribution figures for a project (issue #120).

Each record is {channel, value, source?, period?, metric?, attribution_window?, incremental?}; a row is keyed by (project_id, channel, source, period) so re-uploading overwrites it. Platform figures default to non-incremental (last-touch) unless a record sets incremental=True (a genuine platform geo-lift). Records without a numeric value / a channel are skipped. Returns all of the project’s figures.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.list_platform_figures(project_id)

A project’s platform-reported attribution figures, newest-updated first.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.delete_platform_figures(project_id, *, channel=None, source=None)

Delete platform figures for a project (optionally one channel / source). Returns the number of rows removed.

Return type:

int

mmm_framework.platform.sessions.record_signoff(project_id, approver, *, role=None, note='', assumptions=None)

Append a sign-off approving the current assumption set (issue #110).

Records who approved (approver/role), when, a digest of the assumptions snapshot approved, and a chain hash over the prior record — so the audit trail is append-only and tamper-evident (see verify_signoff_chain()).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_signoffs(project_id)

A project’s sign-offs, newest first.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.verify_signoff_chain(project_id)

Re-derive the hash chain oldest→newest: {intact, n, broken_at?}. A False means a record was altered or removed after the fact.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.GARDEN_STATUSES = ('draft', 'tested', 'published', 'deprecated')

Lifecycle states for a registered garden model.

mmm_framework.platform.sessions.GARDEN_TRANSITIONS: dict[str, set[str]] = {'deprecated': {'draft'}, 'draft': {'deprecated', 'tested'}, 'published': {'deprecated'}, 'tested': {'deprecated', 'draft', 'published'}}

Allowed status transitions (mirrors ALLOWED_TRANSITIONS for experiments). draft -> tested is gated on a passing compatibility report; the human publish gate is tested -> published. Published versions are immutable.

mmm_framework.platform.sessions.DEFAULT_ORG_ID = 'dev-org'

Org id used in the single-tenant dev posture (no auth) — MUST match the _DEV_PRINCIPAL.org_id in api/main.py and AuthSettings.dev_org_id so the agent (resolves org from the project) and the REST layer (resolves org from the principal) agree on where garden models live.

mmm_framework.platform.sessions.resolve_org_id(project_id)

Owning org for a project — the garden registry’s sharing boundary.

Falls back to DEFAULT_ORG_ID in the single-tenant dev posture (or when the auth schema’s projects.org_id column isn’t present yet).

Return type:

str

mmm_framework.platform.sessions.next_garden_version(org_id, name)

Next monotonic version for (org, name) — 1 when none exist yet.

Return type:

int

mmm_framework.platform.sessions.upsert_garden_model(*, org_id, name, version=None, model_id=None, owner_user_id=None, docs=None, manifest=None, source_path=None, base_run_id=None, reference_artifact_path=None, status=None)

Create a new draft (no model_id) or partially update an existing row.

On create, version auto-increments per (org, name) when omitted and the row starts as draft. PUBLISHED versions are immutable — editing a published row raises ValueError (re-publish requires a new version).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_garden_model(*, model_id=None, org_id=None, name=None, version=None)

Fetch by id, or by (org_id, name, version).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.get_latest_garden_model(org_id, name, *, status=None)

Highest-version row for (org, name), optionally filtered by status (e.g. "published" to resolve what a consumer should load).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_garden_models(org_id, *, name=None, status=None, latest_only=False)

Garden models for an org, newest-updated first; optional name/status filter. latest_only collapses to the highest version per name.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.list_garden_versions(org_id, name)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.set_garden_compat_report(model_id, report)

Store the compatibility-suite report for a model (used to gate testing).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.transition_garden_model(model_id, new_status, *, note=None, base_run_id=None)

Validated lifecycle move with an append-only audit trail.

Enforces GARDEN_TRANSITIONS. draft -> tested additionally requires a stored compatibility report whose blocking tiers passed (the automated testing gate); tested -> published is the human publish gate.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.delete_garden_model(model_id)

Delete a garden row. Only draft / deprecated rows are deletable — published history is immutable (deprecate it instead).

Return type:

bool

mmm_framework.platform.sessions.create_data_connection(project_id, name, kind, config)

Persist a reusable data-source connection (no credentials in config).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_data_connections(project_id)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_data_connection(connection_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.get_data_connection_by_name(project_id, name)

Resolve a connection by name within a project (newest wins).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.delete_data_connection(connection_id)

Return type:

bool

mmm_framework.platform.sessions.touch_data_connection_synced(connection_id)

Return type:

None

mmm_framework.platform.sessions.set_data_connection_schedule(connection_id, interval_minutes, *, now=None)

Set (or clear, with None) a connection’s auto-sync interval.

Setting an interval schedules the first run one interval out; clearing it (None) stops scheduled syncs (manual/on-demand still works).

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.list_due_data_connections(now, *, limit=100)

Scheduled connections whose next_sync_at has arrived (across all projects).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.record_data_connection_sync(connection_id, *, status, row_count=None, error=None, snapshot_path=None, now=None)

Record a sync outcome and advance next_sync_at.

On error the next run is backed off (up to 4x the interval, capped so a sub-daily schedule waits at most a day) so a permanently broken connection isn’t retried every interval forever — but never sooner than its own interval.

Return type:

None

mmm_framework.platform.sessions.record_run_metrics(run_id, metrics, *, thread_id=None, project_id=None, artifact_id=None, created_at=None)

Upsert the metrics snapshot for a run. created_at defaults to now; backfill passes the original artifact timestamp so series stay ordered.

Return type:

None

mmm_framework.platform.sessions.list_run_metrics(project_id=None)

Run metrics, OLDEST first (trajectory order).

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_run_metrics(run_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.run_metrics_activity(since_ts)

Deployment-wide fit activity: total fits, fits since since_ts, and the most recent fit time — a lightweight reliability/SLA signal.

Return type:

dict[str, Any]

mmm_framework.platform.sessions.ensure_default_project()

Create the built-in Default Project if absent; return its id.

Return type:

str

mmm_framework.platform.sessions.create_project(name, description=None, org_id=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_projects(org_id=None)

List projects. When org_id is given, scope to that tenant’s projects.

org_id=None preserves the original behavior (all projects + the built-in Default Project) used by single-tenant/dev mode.

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_project(project_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.set_project_meta(project_id, meta)

Merge meta into the project’s onboarding profile (None values delete keys). Returns the updated project, or None for an unknown id.

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.update_project(project_id, name=None, description=None)

Return type:

bool

mmm_framework.platform.sessions.delete_project(project_id)

Delete a project. Its sessions become unassigned (project_id NULL) and its KB documents/chunks are removed. The built-in Default Project cannot be deleted.

Return type:

bool

mmm_framework.platform.sessions.resolve_project_id(thread_id)

The project a session belongs to, falling back to the Default Project.

Never returns None — the knowledge base always has a home.

Return type:

str

mmm_framework.platform.sessions.set_preference(scope, key, value)

Upsert one preference value (stored as JSON) under (scope, key).

Return type:

dict[str, Any]

mmm_framework.platform.sessions.get_preference(scope, key)

Return type:

Any | None

mmm_framework.platform.sessions.list_preferences(scope)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.delete_preference(scope, key)

Return type:

bool

mmm_framework.platform.sessions.create_user(name, email=None, role='analyst')

Return type:

dict[str, Any]

mmm_framework.platform.sessions.list_users()

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.update_user(user_id, *, name=None, email=None, role=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.delete_user(user_id)

Return type:

bool

mmm_framework.platform.sessions.set_project_members(project_id, members)

Replace the project’s member list: [{user_id, role?}].

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.list_project_members(project_id)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_project_branding(project_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.set_project_branding(project_id, branding)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.add_kb_document(project_id, name, path, kind, size_bytes=None, status='pending', meta=None)

Return type:

dict[str, Any]

mmm_framework.platform.sessions.set_kb_document_status(doc_id, status, n_chunks=None, error=None)

Return type:

None

mmm_framework.platform.sessions.list_kb_documents(project_id)

Return type:

list[dict[str, Any]]

mmm_framework.platform.sessions.get_kb_document(doc_id)

Return type:

dict[str, Any] | None

mmm_framework.platform.sessions.delete_kb_document(doc_id)

Return type:

bool

mmm_framework.platform.sessions.add_kb_chunks(document_id, project_id, chunks)

Insert chunk rows. chunks is a list of (chunk_index, text, embedding_blob, dim).

Return type:

int

mmm_framework.platform.sessions.iter_kb_chunks(project_id)

All chunks for a project (id, document_id, chunk_index, text, embedding bytes, dim).

Return type:

list[dict[str, Any]]

History

Host-side run-metrics persistence and history/coverage/priority assembly.

The kernel computes a model-only metrics snapshot at fit time (planning.history.compute_run_metrics); this module is the host half:

• persist_run_metrics: enrich the snapshot with registry calibration status (experiment-backed vs model-only, evidence age) and write the run_metrics row. Registry/DB access stays host-side — kernels never touch the sessions store.

• build_history_series: pivot the stored snapshots into per-channel and portfolio trajectory series for the Performance page (no model loads).

• build_calibration_coverage: channels × evidence tier, with information decay applied at read time.

• build_priorities_payload: the latest EIG/EVOI grid with decay/re-test status recomputed against the registry as of today (closed-form over the stored roi_sd — no model load).

mmm_framework.platform.history.enrich_channel_metrics(metrics, evidence, as_of=None)

Stamp per-channel calibration status from the registry evidence map. Mutates and returns metrics.

Return type:

dict[str, Any]

mmm_framework.platform.history.persist_run_metrics(model_run, thread_id, *, artifact_id=None, created_at=None)

Enrich + persist the metrics snapshot riding a model_run record. Returns the enriched metrics, or None when the run carries no metrics. Never raises (callers treat this as best-effort).

Return type:

dict[str, Any] | None

mmm_framework.platform.history.build_history_series(project_id)

Trajectory series for the Performance page, from run_metrics rows only (oldest first). Channels appearing in any run are included; runs missing a channel simply skip that point (series carry run_id per point).

Return type:

dict[str, Any]

mmm_framework.platform.history.latest_model_run_payload(project_id)

The newest model_run artifact payload in the project — the design endpoints resolve the dataset path + KPI from here (the design engine is data-only; no model load).

Return type:

dict[str, Any] | None

mmm_framework.platform.history.build_calibration_coverage(project_id, *, as_of=None)

Channels × evidence tier with decay applied at read time.

Tier: running (a test is in market or its readout awaits calibration — the live edge of the program), calibrated (experiment-backed, evidence fresh), stale (experiment-backed but decayed past the re-test threshold AND older than the freshness floor), or model_only. Coverage counts calibrated+stale+running-with-evidence as experiment-backed.

Return type:

dict[str, Any]

mmm_framework.platform.history.build_priorities_payload(project_id, *, as_of=None)

The latest stored EIG/EVOI grid with information decay and registry state applied at read time. Returns None when the project has no metrics.

Decay is closed-form over the stored roi_sd + evidence dates, so this never loads a model: stored eig reflects fit time; eig_decayed / retest_due reflect TODAY.

Return type:

dict[str, Any] | None

Runs

MLflow-style run tracking over the existing model_run artifacts.

Every fit already persists a model_run artifact carrying the full normalized spec, the dataset path, and the summary; fit_mmm_model additionally stamps a dataset fingerprint, a spec hash, the parent run id, and a snapshot of the assumption stack at fit time. This module materializes that into a LINEAGE timeline: for each run, what changed vs its predecessor (spec leaves, dataset, assumptions) — the provenance needed to audit the process or write the final report.

mmm_framework.platform.runs.data_fingerprint(path)

Cheap content identity for a dataset file: md5 + size + row count.

Return type:

dict[str, Any] | None

mmm_framework.platform.runs.spec_hash(spec)

Return type:

str | None

mmm_framework.platform.runs.build_run_timeline(project_id=None)

All model runs (newest first), each annotated with its diff vs the chronologically previous run in the same scope.

Return type:

list[dict[str, Any]]

mmm_framework.platform.runs.compare_runs(run_a, run_b)

Per-channel ROI/spend delta between two fitted runs (B relative to A).

Answers the analyst’s most frequent question — “why did this channel’s ROI change since the last refresh?” — from the persisted run_metrics, as a structured per-channel + portfolio delta (rather than eyeballing two reports). Raises ValueError if either run has no stored metrics.

Return type:

dict[str, Any]

mmm_framework.platform.runs.run_timeline_markdown(project_id=None, max_runs=None)

The timeline as markdown — the provenance section of a final report.

max_runs keeps only the most recent N runs (used by the agent tool to bound how much lands in the LLM’s context); None returns the full lineage (used for reports).

Return type:

str

Pacing

Server-side in-flight pacing — actual vs plan (issue #123).

Auto-sources the PLANNED series from the project’s latest saved budget plan and compares it against the stored ACTUAL delivery (the delivery registry), computing per-channel divergence, off-pace flags, and an alert summary — all model-free (no fit load), so it can be served from a request. The expected-outcome delta (which needs the fitted response curves) is left to the check_pacing agent tool / the report pacing= path.

project_pacing is the pure join (unit-testable without a DB / a model); build_project_pacing reads the sessions store.

mmm_framework.platform.pacing.parse_delivery_records(raw, filename='')

Parse an uploaded delivery file (CSV/TSV or JSON) into {channel, period, spend} records (issue #123). Format is chosen by extension, falling back to sniffing the first non-space char ([/{ → JSON). Lenient — bad spend/channel rows are dropped downstream by upsert_delivery.

Return type:

list[dict[str, Any]]

mmm_framework.platform.pacing.project_pacing(plan_payload, delivery_rows, *, threshold=0.1)

Pure: compute the pacing payload + alert digest from a plan + delivery.

Returns the PacingResult.to_dict() shape augmented with available / reason / plan_basis / alert. available is False (with a machine-readable reason) when there is no saved plan or no delivery yet, so the FE can show the right empty state instead of a misleading zero.

Return type:

dict[str, Any]

mmm_framework.platform.pacing.build_project_pacing(project_id, *, threshold=0.1)

Pacing for a project, from persisted data only (no model load).

Auto-sources the planned series from the latest saved budget plan and joins the stored actual delivery. Cheap enough to serve from a request.

Return type:

dict[str, Any]

mmm_framework.platform.pacing.latest_pacing_alert(project_id)

The most recently persisted off-pace alert for a project (issue #123), or None. Written by the background sweep so a planner is alerted without opening the pacing panel.

Return type:

dict[str, Any] | None

mmm_framework.platform.pacing.sweep_pacing_alerts(now=None)

Recompute pacing for every project and persist an off-pace alert artifact (issue #123) — the proactive cadence, so a planner is flagged between fits without opening the panel. Off-pace projects get an upserted pacing_alert artifact; projects that recover have theirs cleared. Returns a digest {scanned, off_pace, persisted, cleared}.

Return type:

dict[str, int]

Scorecard

Recommendation scorecard — predicted vs realized (issue #109).

Nothing builds a CMO’s (and CFO’s) trust in a model faster than watching its past calls come true — or seeing it honestly own the misses. The raw material is already persisted: each fitted run’s per-channel ROI (the prediction at recommendation time) and the experiment registry’s calibrated readouts (the realized incremental return). This joins them, per channel, into an accountability view: predicted (with its credible interval) vs realized, the error, and whether the realized value landed inside the predicted interval — so the model’s honesty (interval calibration) is auditable over time.

All from persisted data — no model load. project_scorecard_rows is the pure join; build_project_scorecard reads the sessions store.

mmm_framework.platform.scorecard.project_scorecard_rows(estimands, experiments)

Join persisted MMM contribution_roi predictions to realized experiment readouts. Pure — takes build_project_estimands() output + registry experiment dicts. Returns {rows, calibration, n_recommendations}.

Each row pairs a channel’s realized experiment ROAS with the ROI the model predicted for it — preferring the experiment’s recommending_run_id (the run that recommended the test), else the latest run that estimated the channel. in_interval records whether the realized value landed inside the predicted credible interval (the calibration signal).

Return type:

dict[str, Any]

mmm_framework.platform.scorecard.build_project_scorecard(project_id)

The recommendation scorecard for a project, from persisted data only.

Return type:

dict[str, Any]

Triangulation

Server-side triangulation join — MMM × experiment × platform (issue #119).

The reconciliation engine (mmm_framework.reporting.triangulation) already puts the three independent estimates of a channel’s return next to each other and classifies convergent / divergent / platform-inflated. This module wires it to persisted data so the Oracle / an endpoint can pull the panel WITHOUT reloading or re-fitting a model:

• MMM sources come from the per-channel contribution_roi estimand rows already persisted on each model_run artifact (via build_project_estimands() — no model load);

• experiment sources come from the experiment registry (sessions.list_experiments(), calibrated + completed readouts);

• platform figures are optional (an inline dict until the platform-ingestion follow-up, #120, adds a persistence slot).

project_triangulation_sources is the pure join (unit-testable without a DB); build_project_triangulation reads the sessions store.

mmm_framework.platform.triangulation.project_triangulation_sources(estimands, experiments, *, kpi=None, run_id=None, platform=None)

Reconcile persisted MMM estimands + registry experiments (+ platform).

Pure: takes the build_project_estimands() output + a list of experiment registry dicts and returns a serializable payload. No DB / no model.

Return type:

dict[str, Any]

mmm_framework.platform.triangulation.parse_platform_records(raw, filename='')

Parse an uploaded platform-figures file (CSV/TSV or JSON) into {channel, value, source?, period?, metric?, attribution_window?, incremental?} records (issue #120). Format is chosen by extension, falling back to sniffing the first non-space char. A CSV is a table with a channel + value column (+ optional source/metric/ attribution_window/incremental/period); JSON may be a list of such records or a {channel: value} / {channel: {value, ...}} map.

Return type:

list[dict[str, Any]]

mmm_framework.platform.triangulation.platform_dict_for_project(project_id)

Reduce a project’s stored platform figures to the one-per-channel {channel: {value, metric, attribution_window, incremental}} dict the triangulation engine’s platform input expects (issue #120).

Platform figures are return ratios (ROAS), not additive spend, so they are NOT summed; the most-recently-updated figure per channel wins (rows arrive newest-first from list_platform_figures()).

Return type:

dict[str, Any]

mmm_framework.platform.triangulation.build_project_triangulation(project_id, *, kpi=None, run_id=None, platform=None)

Triangulation panel for a project, joined from persisted data only.

Reads the project’s persisted MMM contribution_roi estimands + the experiment registry + the stored platform-attribution figures (issue #120) and reconciles them channel-by-channel. No model is loaded, so this is cheap enough to serve from a request. An explicit platform dict overrides the stored figures.

Return type:

dict[str, Any]

Estimands

Project estimand aggregation for the Performance page.

Turns the estimand rows persisted on each model_run artifact (see agents.fitting / agents.estimand_rows) into comparability clusters: one cluster per (estimand, KPI) so that the same estimand measured on the same KPI by different models sits side-by-side, while a model on a different KPI lands in its own cluster and is never silently compared. Two ROI estimands of the same statistical kind but different methodology (contribution_roi vs counterfactual_roi) stay distinct clusters — they are different numbers.

The labeling / reference-value / evidence logic mirrors the report’s EstimandsSection exactly (reporting/sections.py) so the dashboard and the generated report tell the same story:

• reference value: 1.0 for ratio kinds (ROI / ROAS / x / multiple), else 0.0

• evidence vs the credible interval: lower > ref -> “strong”; upper < ref -> “below”; otherwise “uncertain”; missing/unsupported -> “na”.

group_estimands is pure (no DB) and unit-tested; build_project_estimands reads the sessions store.

mmm_framework.platform.estimands.estimand_label(name)

Human label for an estimand name (title-cased fallback for unknowns).

Return type:

str

mmm_framework.platform.estimands.is_ratio_kind(kind, units)

A ratio estimand (reference 1.0) — mirrors EstimandsSection._is_ratio_kind.

Return type:

bool

mmm_framework.platform.estimands.classify_evidence(*, status, mean, lower, upper, reference)

Evidence label vs the no-effect reference: strong / below / uncertain / na.

Return type:

str

mmm_framework.platform.estimands.group_estimands(runs)

Cluster per-run estimand rows into (estimand, KPI) comparability groups.

Return type:

dict[str, Any]

Parameters

runs

One dict per fitted run with keys run_id, label, model_kind, model_key, kpi, created_at and estimands (a list of rows as produced by mmm_framework.agents.estimand_rows.evaluate_estimand_rows()).

Returns

dict with runs (incl. is_latest_for_model for the default selection), kpis, and groups (each a comparability cluster). Pure; no I/O.

mmm_framework.platform.estimands.build_project_estimands(project_id)

Read the project’s model_run artifacts and group their persisted estimand rows. Runs fitted before estimand persistence (or with no estimands) are skipped; run the backfill (python -m mmm_framework.platform.backfill --what estimands) to populate them.

Return type:

dict[str, Any]

Backfill

Backfill run snapshots for model runs fitted before a feature existed.

Two backfills over the same model_run artifacts:

• run_metrics — for every artifact without a run_metrics row: rebuild the panel from the recorded spec + dataset, deserialize the saved model, run compute_run_metrics, and persist with the artifact’s original timestamp so trajectory series stay ordered.

• estimands — for every artifact whose payload has no estimands: load the model, realize its declared/default estimands, and write the rows (+ model_kind) back onto the artifact payload so the Performance estimands view can show pre-existing models.

• evidence — for every MMM artifact whose payload has no channel_evidence: load the model and compute the per-channel evidence tier + identifiability flag so the live Performance/Estimands dashboard carries the same chip the report does (issue #124).

Skip-and-report anything unrecoverable (empty model dirs, moved datasets) — old runs saved on other machines often are.

Usage:

python -m mmm_framework.platform.backfill [–project ID] [–dry-run]

[–max-draws 200] [–what {metrics,estimands,evidence,all}]

mmm_framework.platform.backfill.backfill_run_metrics(project_id=None, *, dry_run=False, max_draws=200)

Returns a per-run status report: {run_id, status, detail} with status ∈ {done, exists, skipped, error, would-run}.

Return type:

list[dict[str, str]]

mmm_framework.platform.backfill.backfill_estimands(project_id=None, *, dry_run=False, random_seed=42)

Populate payload['estimands'] (+ model_kind) on every model_run artifact that lacks it, so the Performance estimands view covers models fitted before estimand persistence existed. Returns a per-run status report with status ∈ {done, exists, skipped, error, would-run}.

Return type:

list[dict[str, str]]

mmm_framework.platform.backfill.backfill_evidence(project_id=None, *, dry_run=False)

Populate payload['channel_evidence'] (per-channel evidence tier + identifiability flag) on every MMM model_run artifact that lacks it, so the live Performance/Estimands dashboard shows the same evidence chip the report does for models fitted before this snapshot existed (issue #124). Returns a per-run status report with status ∈ {done, exists, skipped, error, would-run}.

Return type:

list[dict[str, str]]

mmm_framework.platform.backfill.main()

Return type:

None

Backup

Online backup / restore for the SQLite sessions + auth store.

The entire platform state (sessions, auth users/tokens, run_metrics, langgraph checkpoints) lives in one SQLite file with no backup tooling — a single disk loss destroys everything. This adds a CONSISTENT online backup (SQLite’s backup API, safe while the app is running, WAL-aware) and a restore, plus a CLI:

python -m mmm_framework.platform.backup backup  /backups/sessions-2026-06-25.db
python -m mmm_framework.platform.backup restore /backups/sessions-2026-06-25.db

This is the stop-gap for the single-SQLite risk until the Postgres migration (action plan P1a). See technical-docs/disaster-recovery.md.

mmm_framework.platform.backup.backup_db(dest, *, db_path=None)

Write a consistent snapshot of the live DB to dest. Returns dest.

Return type:

Path

mmm_framework.platform.backup.restore_db(src, *, db_path=None)

Restore the DB from a backup at src into the live path. Returns target.

Return type:

Path

mmm_framework.platform.backup.main(argv=None)

Return type:

None