Standalone OG-CORE Ingestion Pipeline Backend

[Adityakushwaha2006]

• Closes [Feature] Standalone OG-Core modelling layer (Backend Ingestion Layer) #470

Summary

What changed: adds the OG-Core run pipeline, the part that runs an installed calibration and serves its results , fully end to end. Thirty ogc endpoints(as per specific contract) for cases, runs, parameters, execution, results, the analysis tables, tax uploads, the parameter form schema, and case backup and restore. All the model work is done by a permanent worker script that runs OG-Core in its own country specific environment; MUIOGO never imports ogcore for in process runs (issue with #471), All calls are sub processes.

Why: This is the layer that runs the entire computation for OG-CORE : the frontend can now create a case, solve a baseline and a reform, and read the results back. It is the cross-environment run layer that was mentioned under #470.

---

Architectural Explanation

flowchart TB
    FE["Frontend (browser)"] -->|"HTTP /ogc/..."| RT["OGCoreRunRoute<br/>web layer"]
    RT --> JOB["RunJob<br/>one solve at a time, queue, cancel"]
    RT --> ST["OGCoreCase<br/>case and run files on disk"]
    JOB --> RUN["OGRunner<br/>spawns and watches the worker"]
    RUN -->|"looks up python_path"| REG["Installed registry (from #486)"]
    RUN -->|"spawns with the calibration's own python"| W["ogc_worker.py<br/>the only code that imports ogcore"]
    W --> OG["OG-Core, in its own environment<br/>solve, then read its own pickles,<br/>write plain JSON results"]
    RT -->|"results endpoints read plain JSON"| DISK["run folder on disk"]
    W --> DISK

Loading

How a Run happens :

1. The user creates a case, which is tied to one installed country. Inside it they create a baseline run and, later, one or more reform runs. Parameters are entered per run.
2. The user clicks run. The route validates the request and hands it to the job layer, then answers the browser immediately. (Since the solve takes quite long)
3. The job layer runs two safety checks (a reform must have a completed baseline, and must use the same model dimensions as that baseline), then claims the single execution slot. If another solve is already running, this one gets queued
4. The launch layer looks up the calibration's Python in the registry(containing the env paths) and spawns the worker with it. It streams the worker's output, pulls out the iteration count for a live progress readout, and enforces a maximum run time.
5. The worker builds the parameters, solves the steady state and (if asked) the transition path, reads its own result pickles, and writes them back as plain unpickled JSON.
6. The frontend polls for status (frontend side execution - preferably long async poll), and once the run is completed, reads the results through the result endpoints (both consolidated and atomic endpoints kept in)

Additional Things to note:

• This architecture fully supports Multi Industry Runs (M>1) end to end - it requires no further changes.
• One permanent worker script. There is no code generation per run. What changes between runs is the data in the run folder, never the code.
• Completion comes from the process exit code plus the terminal status the worker wrote, never from a file merely appearing and never from scraping printed output.
• The parameter form is read live from the installed calibration's own defaults file, so it always matches the version the user actually has.
• Cancel and the maximum run time both kill the whole process tree and save nothing. A run is only ever read once its status says completed.
• The solve runs in a background thread on the server, not tied to the browser, so a refresh does not stop it. On reload the run still shows running via getRuns/getRunStatus and the frontend just resumes polling; if the whole server was restarted mid-run, getRunStatus repairs it to failed ("interrupted by restart"). Nothing is lost either way.

To Obtain the entire case-run tree on the frontend:

1. getCases (GET) -> every case with its summary (casename, country_id, description, modified_at, has_results). This is the top level of the navigation.
2. For each case, getRuns (POST, {casename}) ->that case's baseline and reforms, each with live status. Loop this over the list from step 1 to get every run under every case.
3. Drill-down into a run as the user clicks: getParams, getResults, getRunStatus.

---

Validation

• Tests added/updated
  Automated suites across the build: driving 12 real model solves through the full stack (steady state and transition path, baseline and reform, plus the queue, a cancel, and a retry after cancel). Every guard, every error shape, and every result endpoint was exercised. Full Parity achieved with direct OGC runs on sample data - proving that there is prevalent logical/computational bug in the processing pipeline. All tests have been conducted on the base model.

• Manual verification steps documented, with evidence where relevant (attached below)

Live Manual Verification :

Live session against the running backend. Every call is a real HTTP request.

Reference : installed base metadata in installation registry json

1. A real model solve, completed through the API

Shows a run started through /ogc/run, polled through to completed, with the live
stage and iteration count climbing as it solved. This is a full transition-path solve
of OG-Core, launched by the pipeline.

2. The parameter form schema

getParameterSchema returning all 129 parameters, each with its label, grouping,
default, and range, read live from the installed calibration.
(Verified everything is returned )

3. Environment available, execution slot free then locked

Above: While Running log and Status Check

Two terminals side by side. Before the run, the installed environment is listed and
the run is pending (the slot is free). During the run, the same environment is still
listed and the run is now running (the slot is occupied). The environment path is
always available.

4. Two runs: one running, one queued

• call a run while another is running

• Queue State

A second run scheduled while the first is still solving comes back as queued, and its
status shows pending with a Queued stage. This is the one-solve-at-a-time rule in
action.

5. Cancel, and the queue advancing on its own

Cancelling the running solve returns cancelled, the run then shows failed with the
reason "Cancelled by user.", and the queued run moves to running by itself. This shows
cancel works and the queue advances automatically.

6. Consolidated results (baseline vs reform)

• Terminal Output visually confirmed

getResults returning the dashboard object: the years, each variable's baseline path,
its reform path, and the percent difference, plus the steady-state values.

7. Analysis tables and the CSV download

• Terminal Output visually confirmed

The analysis tables (macro, inequality, gini, time series, revenue decomposition)
returning real rows, computed live from the saved results, and the macro table
downloaded as a CSV file.

8. A prior run fetched from a fresh terminal

• Terminal Output visually confirmed

A run that finished earlier, fetched from a brand-new terminal with no shared session:
its status, its saved parameters, and its results all come back. This shows run
results are saved durably and are readable later, independent of the session that
created them.

---

Files

New:

• API/Classes/OGCore/OGCoreCase.py (cases and runs on disk)
• API/Classes/OGCore/ogc_worker.py (the OG-side worker, the only code that imports
  ogcore)
• API/Classes/OGCore/OGRunner.py (spawns and supervises the worker)
• API/Classes/OGCore/RunJob.py (one solve at a time, queue, cancel)
• API/Classes/OGCore/OGResults.py (reads the worker's plain JSON into the dashboard
  shape)
• API/Classes/OGCore/OGTables.py (drives the worker's short table and check calls)
• API/Classes/OGCore/OGSchema.py (builds the parameter form metadata)
• API/Routes/OGCore/OGCoreRunRoute.py (all the endpoints above)

Changed:

• API/Classes/Base/Config.py (the cases storage path)
• API/app.py (registers the new blueprint)

Checklist

• Change is scoped — no unrelated refactors
• Branched from main; PR targets EAPD-DRB/MUIOGO:main
• Docs updated for any setup, workflow, or architecture change (the design is documented in this PR body and under [Feature] Standalone OG-Core modelling layer (Backend Ingestion Layer) #470; no in-repo docs were changed)

Additional Info:

• In-depth system Architecture for the ingestion layer in this PR

flowchart TB
    subgraph BROWSER["Browser (frontend)"]
        FE["JavaScript<br/>sends HTTP requests<br/>renders results"]
    end

    subgraph FLASK["MUIOGO Flask app, its own environment, port 5002"]
        subgraph ROUTE["OGCoreRunRoute.py<br/>WEB LAYER"]
            R1["receives HTTP request"]
            R2["validates structure<br/>fields present, names safe,<br/>cross-site guard"]
            R3["passes raw values<br/>as function args"]
            R4["wraps return dict<br/>in jsonify"]
        end

        subgraph JOB["RunJob.py<br/>JOB LAYER"]
            J1["runs the two guards"]
            J2["claims the run,<br/>queues if one is solving"]
            J3["tracks status,<br/>cancel, timeout"]
        end

        subgraph RUNNER["OGRunner.py<br/>LAUNCH LAYER"]
            L1["looks up python_path<br/>in the registry"]
            L2["spawns the worker<br/>with a clean env"]
            L3["streams output to log,<br/>extracts iteration count"]
            L4["kills the process tree<br/>on cancel or timeout"]
        end

        subgraph STORAGE["OGCoreCase.py<br/>STORAGE LAYER"]
            S1["case CRUD, genData.json"]
            S2["run CRUD, run_meta.json"]
            S3["params read/write,<br/>ogcParams.json"]
            S4["name safety guard"]
        end
    end

    subgraph REG["Installed registry (#486)"]
        RG1["country_id to<br/>python_path"]
    end

    subgraph DISK["Disk: DataStorage/OGCore/casename/"]
        D1["genData.json"]
        D2["res/runname/run_meta.json"]
        D3["res/runname/ogcParams.json"]
        D4["res/runname/run_status.json"]
        D5["res/runname/SS + TPI pickles"]
        D6["res/runname/results_ss.json<br/>results_tpi.json"]
    end

    subgraph OGENV["OG environment: the calibration's own .venv, SEPARATE PROCESS"]
        W["ogc_worker.py<br/>the only code that imports ogcore"]
        OG1["Specifications<br/>defaults + layers + overrides"]
        OG2["runner: SS solve then TPI solve"]
        OG3["output_tables, validator"]
    end

    FE -->|"HTTP POST/GET"| R1
    R1 --> R2 --> R3
    R3 -->|"run endpoints"| JOB
    R3 -->|"CRUD endpoints"| STORAGE
    JOB --> RUNNER
    RUNNER -->|"python_path"| REG
    RUNNER -->|"spawn subprocess"| W
    W --> OG1 --> OG2
    W -->|"reads own pickles,<br/>writes plain JSON"| D6
    OG2 -->|"writes pickles"| D5
    W -->|"stage updates,<br/>terminal status"| D4
    STORAGE -->|"file I/O"| DISK
    R3 -->|"results endpoints<br/>read plain JSON only"| D6
    JOB --> R4
    R4 -->|"HTTP response JSON"| FE

Loading

[Adityakushwaha2006]

Hey @error9098x ,
Before i ask Marcelo and Alfonso for a review, ill wait for a confirmation from your side on double checking all API outputs and processing so we can be sure that this is exactly how we want everything laid out,

@utsinboots Would appreciate if you could test the same aswell !

[utsinboots]

Tested PR 498 on Windows 11 with uv run python API/app.py
Ran multiple baseline and reform solves through the API: create case create run run poll status getRunStatus getResults baseline vs reform results CSV export; also:

• one solve-at-a-time queueing: second run queues behind the first and auto-starts once it finishes
• active-run cancellation: reaches failed with "Cancelled by user.", and the queue advances on its own
• case backup and restore: zip preserves exact case/run data; restore correctly refuses to overwrite an existing case
• server restart recovery: if the server crashes or restarts mid-solve, the run doesn't stay stuck showing "running" forever; the next status check notices there's no process behind it anymore and automatically marks it failed with a clear reason ("Run was interrupted by an application restart.")
• invalid-input handling: missing fields, path traversal, wrong types, unknown runs, malformed JSON all return clean errors, no crashes.

All the above tests PASS as described. I found two issues while testing:

1: Cancelling a queued run leaves it permanently stuck as pending
Start run A so it occupies the execution slot.
Start run B so it is queued behind run A.
Cancel run B while it is still queued.
Allow run A to finish.
Check the status of run B.
Run B remains:

{
  "run_state": "pending",
  "run_stage": null
}

It never transitions to running, completed, failed, or another terminal state.

2: downloadResults mislabels a solo run export as "Baseline"
Calling downloadResults with only base_run set exports a single run. However, if the run supplied through base_run is actually a reform run, for example: (e.g. base_run=reform1), the CSV still labels every row/column "Baseline".
Since the label seems hardcoded to whichever run occupies the base_dir slot rather than derived from the run's name or type.

Let me know if I missed anything and I'll give it another go. Also, curious to know, how do I test the automated test suites you described in the PR?

[Adityakushwaha2006]

Hey @utsinboots , thanks for catching those , fixed them in the latest commit.
For the test suite : use uv run pytest
or for specific sets use uv run pytest 'path'

Certain changes pertaining to full correctness with the 503 fixes remain and are not folded in currently. Ill work on those once 503 is in main.

Lemme know if you find anything else that needs to be fixed :)

[utsinboots]

Tested the latest changes Windows 11
Automated tests: Focused OG-Core suite: 43 passed; Full test suite: 92 passed
Queued cancellation: run B changed from pending / Queued to failed, with "Cancelled by user.". It remained failed and did not start after run A completed.
Solo reform labeling: Completed baseline and reform TPI solves and tested reform1 as a solo run.
downloadResults?base_run=reform1 generated a valid CSV.
CSV rows were labeled with the actual run name, for example GDP ($Y_t$) (reform1).
The solo reform CSV contained no incorrect Baseline labels.
getMacroTable used (reform1).
getIneqTable, getGiniTable, and getTimeSeriesTable used reform1 rather than Baseline.
Baseline-versus-reform comparison behavior remains valid.
Both previously reported issues are fixed.

[Adityakushwaha2006]

@error9098x , implemented some of the changes you recommended , thanks for those.

On the validation of Params - it currently retains its original state of atomic validation of bounds, type etc. I tried out the approach you had mentioned to check conflicting attributes and validate them , but the recommended method did not work out.

We can possibly enforce hard correlation rules but that would be a stretch, decided against doing that.

[error9098x]

@error9098x , implemented some of the changes you recommended , thanks for those.

On the validation of Params - it currently retains its original state of atomic validation of bounds, type etc. I tried out the approach you had mentioned to check conflicting attributes and validate them , but the recommended method did not work out.

We can possibly enforce hard correlation rules but that would be a stretch, decided against doing that.

Thanks for explaining that. That makes sense, and I don't see this as a blocker for the PR. I mainly wanted to confirm the behavior because these invalid shapes can pass the atomic validation and fail later inside OG-Core with a low-level error. Since you are keeping validation focused on individual bounds, types, and formats, I will treat the cross-parameter checks as a known limitation rather than a required change here.

[Adityakushwaha2006]

Conducted extensive functional tests by deployment on an active server with only API based I/O
models :

• Base Vanilla OGC
• Country Calibrated OGC-ETH

The tests that were covered:

Server health

	Checked
A1	The server starts and answers a request
A2	Installed calibrations are listed with their state and interpreter
A3	The catalogue loads live and shows the right state for each country
A4	The catalogue still loads from cache with no network

Installing calibrations

	Checked
B1	The base model install left a working folder, environment and interpreter
B2	The country install recorded ogeth rather than ogcore, with its own interpreter
B3	Two different countries install at the same time
B4	A second install of a country already installing is refused
B5	Cancelling an install stops it and clears the part downloaded folder
B6	A check only refresh reports both commit ids and harms nothing
B7	A source can be checked before adding it: a real folder, a wrong folder, a GitHub link
B8	Installing from a GitHub address works
B9	Registering a calibration from a folder already on disk works, in place
B10	A failed install keeps both its reason and its log
B11	Retrying after a failure is accepted and leaves no stale error behind
B12	The catalogue reads not installed, installing, installed and failed at the right moments
B13	An available update is offered and the calibration stays usable
B14	Unregistering removes the record but leaves the files alone

Cases

	Checked
C1	A case is created and appears both in the list and on disk
C2	A case cannot be made for a calibration that is not installed
C3	Editing a case keeps its runs
C4	The country of a case cannot be changed
C5	The active case can be set, read back and cleared
C6	A case reports results only once a run has completed, and not if it failed
C7	Each case reports its own calibration
C8	A case can only be deleted when it is the active one

Runs and parameters

	Checked
D1	A baseline run is created and listed as pending
D2	A case can only have one baseline
D3	A reform is created against its baseline and records it
D4	A reform cannot name a baseline that does not exist
D5	Parameters save and read back unchanged
D6	A reform starts from the baseline's values and can then differ without touching it
D7	The base model's parameter form carries titles, ranges and sections
D8	A country's form keeps all of that and shows the country's own values
D9	Deleting a reform leaves the case and its baseline alone
D10	Deleting the baseline removes the case, and only when it is the active one

Running a solve

	Checked
E1	A solve starts and reports itself as running
E2	The log grows and its last line changes while the solve runs
E2b	The iteration count climbs through the transition path, with the distance shrinking
E3	The worker is a separate process running under the calibration's own environment
E4	A solve completes, writes its results files and leaves no worker behind
E5	The same run cannot be started twice
E6	A reform cannot run before its baseline has finished
E7	A reform must match the baseline on S, T, J, M and I
E8	A finished run can be run again and still has readable results afterwards
E9	The start and finish times survive on disk
E10	A country solve records its country package, its version and the environment's ogcore

The queue

	Checked
F1	A second run is queued rather than started, and only one solve runs
F2	The queue moves on by itself with no further request
F3	Three runs sent together are all accounted for and all finish
F4	Queued runs start in the order they arrived

Cancelling

	Checked
G1	Cancelling kills the worker and every child it started
G2	Cancelling frees the slot and the queued run takes it
G3	Cancelling a queued run ends it instead of leaving it pending forever
G4	Cancelling something that is not running is refused
G5	A cancelled run says why it failed, not just that it did

Interruption and recovery

	Checked
H2	A run interrupted by a crash is repaired on the next start
H3	A worker that survived the crash is killed on the next start
H4	A completed run and its results survive a restart untouched

Runs and installs together

	Checked
I1	A run is refused while its own calibration is being installed
I2	An update is refused while a run is using that calibration
I3	Reinstalling over a calibration in use is refused
I4	Registering a local copy over one in use is refused
I5	A first install of a different country is allowed

Results and analysis tables

	Checked
J1	Steady state variables come back, with sensible values
J2	Asking for named variables returns only those
J3	The inequality and Gini tables return real rows
J4	Wealth moments work with no survey data: nine rows, no empty column
J5	Wealth moments show the supplied data beside the model values
J6	The macro table refuses clearly when there is no transition path
J7	A single run's table is labelled with that run's name, not baseline
J8	Comparing two runs reports the change, and labels both when levels are asked for
J9	The consolidated view assembles the years, series, steady state and labels
J10	Time series and revenue decomposition return rows, and the stationarized option works
J11	A solve in progress and one never started are told apart
J11b	The granular variable reads do not make that distinction
J12	Transition path variables come back, and are refused for a steady state only run
J13	The steady state comparison needs both runs and says so
J14	Parameter checking catches out of range and wrongly shaped values by name
J15	Two runs read side by side without either leaking into the other
J16	A country run records its provenance

Export, backup and restore

	Checked
K1	Results download as a CSV carrying the same numbers the API returns
K2	A case backs up to a zip with its runs, parameters and results
K3	A deleted case restores with everything intact and its results still load
K4	Restoring over a case that still exists is refused, overwriting nothing
K6	Tax parameters upload and are reported back, and a bad file is rejected cleanly

Refusals and bad input

	Checked
L1	Malformed JSON is rejected
L2	A missing field is named in the refusal
L3	A wrong type is rejected
L4	Path traversal in a name is rejected
L5	Unknown cases and runs return not found
L6	Parameters cannot be changed for a running or a queued run
L7	A case with a run going cannot be deleted
L8	Oversized uploads are refused with the limit named

---

The team can use these as templates for functional testing, feel free to extend this list if ive missed anything important, we can cover the same to check every bit of the Frontend-Backend bridging and functionality.

Another update is that the current PRs pertaining to the frontend made by Aviral are already based on the Backend Infra in this PR, functionality is being tested while frontend is being developed as well, hence solidifying it further.

[Adityakushwaha2006]

@marcelolafleur Ready for review!

[marcelolafleur]

Thanks @Adityakushwaha2006, this is careful work. I read all of it: the process boundary is clean (MUIOGO itself never touches ogcore), runs recover properly after a crash or restart, and the tests exercise the real queue and shutdown paths. Approving.

One fix before merge: restoreCase unpacks the backup straight into the final case folder. If extraction fails halfway (disk full, or a file name Windows refuses) the user is left with a broken half-case, and restoring again is blocked because the case now "exists". Unpack to a temporary folder and move it into place only once everything succeeded.

[Adityakushwaha2006]

@marcelolafleur All done now :)

[error9098x]

@Adityakushwaha2006, is it possible to store cases using a country-specific directory structure?

  cases/<country_id>/<casename>

Currently, cases are stored as:

  cases/<casename>

This can cause name collisions when different country workspaces use the same case name. The country-aware structure may need to be reflected in storage, API lookups, queue handling, workers, and results handling.

[Adityakushwaha2006]

@error9098x Ive implemented the requested change , I could see the problem you were running into , tried to make a way around it but then decided to implement the recommendation itself.

A note for you , and others , this change breaks away from the API contract we had laid out , and also the access point-save point operations in some cases internally, that might already be in the frontend's code that has been merged.

Ive ran my functional tests and my pytests on this to verify, need to note that every endpoint in the affected tree would ask for country id aswell. The diff for the particular commit would help pin point exactly where the API follow-through changes internally , and that'd be the first thing i'd recommend changing in the frontend.

Let me know if this works fine , and if you need anything else.