Scenario workspaces and IDE protocol

The abacus.scenarios namespace exports the workspace, job and protocol types used by external integrations. Application wrappers are separate projects. The library keeps the estimator and scenario operation gates in force.

Workspace objects

load_planner_run_context(results_dir) reconstructs the fitted run context. WorkspaceService(run_context, *, store=None, job_runner=None) uses a local WorkspaceStore and SynchronousScenarioPlannerJobRunner by default.

WORKFLOW_STATUSES contains draft, in_review, approved and archived. normalize_workflow_status(value) normalises a workflow label; iso_now() returns a UTC timestamp. Workflow metadata does not establish statistical approval of the estimator or a causal interpretation.

Persistence and revision checks

WorkspaceStore(results_dir) stores <workspace_id>.json and, when supplied, <workspace_id>.manifest.json under scenario_planner/workspaces/. Individual JSON writes use a temporary file and replacement through atomic_write_json. This does not make multiple file writes or a read-check-write cycle one atomic transaction.

WorkspaceService.save_workspace(workspace, *, action="save", changed_scenario_ids=None, owner=None, job=None, expected_updated_at=None) returns the saved Path. It appends a revision record and updates the supplied workspace’s timestamp and metadata to the saved state.

For an existing workspace, pass the timestamp from the version you loaded as expected_updated_at. A mismatch raises WorkspaceConflictError, whose attributes include workspace_id, expected_updated_at, actual_updated_at and latest_revision_id. Omitting the argument skips this stale-write check. Reload and reconcile a conflict before saving again.

The check and write do not acquire a cross-process transaction lock. External applications must serialise concurrent writes when they need that guarantee. Low-level WorkspaceStore.save(...) also bypasses the service’s revision and conflict handling.

The following function updates an existing workspace. Its arguments identify a fitted run and an existing workspace; it does not create a demo run.

from abacus.scenarios import WorkspaceService, load_planner_run_context


def rename_workspace(results_dir, workspace_id, name):
    service = WorkspaceService(load_planner_run_context(results_dir))
    workspace = service.load_workspace(workspace_id)
    expected_updated_at = workspace.updated_at
    updated = service.update_workspace_metadata(workspace, workspace_name=name)
    return service.save_workspace(
        updated,
        action="rename",
        expected_updated_at=expected_updated_at,
    )

Background jobs

A job runner implements get_job(job_id), get_result(job_id), update_job(job), submit(...) and run(...). submit and run take keyword-only job_type, run_id, workspace_id and callable fn, plus optional scenario_id, cache_key and metadata. submit returns PlannerJob; blocking run returns JobExecutionResult. get_result returns None until a terminal result is available. Terminal states are completed and failed; queued work can also be queued or running.

submit_draft_evaluation(workspace, draft) returns an updated workspace and a job. Once terminal, apply_draft_evaluation_job(workspace, job_id=...) returns an updated workspace, job and optional cache information. The corresponding sensitivity and export pairs are submit_sensitivity_sweep / apply_sensitivity_sweep_job and submit_export_bundle / apply_export_bundle_job.

Apply results explicitly and then save the returned workspace. Polling a job does not persist its result to a workspace. A failed evaluation job records failure without adding a successful evaluation. The caller must reconcile workspace or draft changes that occurred while the job ran; do not apply a stale result to a newer draft without checking its identity and specification. The default runner’s job registry is in memory and does not resume execution after a process restart.

Protocol envelope

PlannerIdeService() keeps one loaded run context per service instance. handle_request(envelope) accepts a dictionary and returns a dictionary. process_request(service, request_line) accepts JSON text and returns JSON text. Import both from abacus.scenarios.

import json

from abacus.scenarios import PROTOCOL_VERSION, PlannerIdeService, process_request

service = PlannerIdeService()
request = {
    "protocol_version": PROTOCOL_VERSION,
    "request_id": "ping-1",
    "action": "ping",
    "payload": {},
}
response = json.loads(process_request(service, json.dumps(request)))
assert response["ok"] is True

The current PROTOCOL_VERSION is 1. Successful responses contain protocol_version, request_id, ok: true and payload. Failures contain ok: false and an error object with code, message and optional detail. PlannerIdeProtocolError.to_error_payload() provides that error representation.

Run-context loading returns run_context. Workspace operations return the workspace and associated context or comparison data. open_workspace and create_default_workspace use UI payloads; create_workspace returns the workspace’s persistence dictionary. get_workspace and draft operations return a workspace document with evaluation summaries and separate comparison data. Do not assume these payload variants have identical nested shapes.

A save_draft response can have ok: true with non-empty validation_errors; in that case the proposed draft was not saved. Check that list. Envelope error codes include invalid_json, invalid_request, invalid_protocol_version, unknown_action, file_not_found, bridge_state_error and internal_error. Unexpected exceptions, including workspace conflicts, use internal_error. The bridge does not expose a caller-supplied revision token; use the Python service when the client needs the timestamp check described above.

For a process bridge, python -m abacus.scenario_planner.ide_service reads one JSON envelope per non-empty stdin line and writes one response per line until stdin closes. This is distinct from the recipe CLI in abacus.scenarios.

Related guides

• Scenario Planner API
• Scenario Python workflow