frbus_shockWhat happens to US GDP and unemployment if the Fed raises rates by 100 basis points?
hank_shockHow does a 25 basis point US rate cut hit households across the wealth distribution?
+
+ Slow calls run as jobs. The hosted server sits behind
+ a transport that abandons any request after 150 seconds, and
+ score_reform over its default five-year window takes
+ longer. Ask for it normally — the model should reach for
+ start_job and then get_job_result, which
+ hands the work to a 30-minute worker and polls it. Running locally
+ there is no such limit and no job step.
+
Terminal setup
diff --git a/integration/README.md b/integration/README.md
index 8a19aa2..02ad555 100644
--- a/integration/README.md
+++ b/integration/README.md
@@ -212,8 +212,12 @@ benefits the MCP server).
## MCP server
Runs over stdio via `python -m policyengine_macro.mcp_server`, exposing
-26 tools:
+28 tools:
`score_reform` (a PolicyEngine reform through a chosen macro model),
+`start_job` and `get_job_result` (run a slow tool in the background and
+collect it — the hosted transport abandons any request after 150s, and
+`score_reform` over its default five-year window needs longer; there is
+no such limit locally, and no job backend, so call tools directly here),
`frbus_shock_incidence`, `hank_shock_incidence` and
`svar_inflation_incidence` (macro-to-microsimulation incidence overlays),
`dynamic_reform_impact` (the OG-UK overlay dynamic score; local-only —
diff --git a/integration/modal_app.py b/integration/modal_app.py
index 2e96f74..a91851c 100644
--- a/integration/modal_app.py
+++ b/integration/modal_app.py
@@ -2,8 +2,8 @@
modal deploy integration/modal_app.py
-Serves the FastMCP instance from `policyengine_macro.mcp_server` (26 tools:
-list_model_capabilities, get_model_status, recommend_model,
+Serves the FastMCP instance from `policyengine_macro.mcp_server` (28 tools:
+start_job, get_job_result, list_model_capabilities, get_model_status, recommend_model,
format_score_report, score_reform, obr_shock, list_reform_variables, frbus_shock,
frbus_list_variables, frbus_summary, frbus_shock_incidence, hank_shock,
hank_summary, hank_shock_incidence, forecast_uk,
@@ -298,6 +298,37 @@
pe_data_volume = modal.Volume.from_name("policyengine-macro-pe-data", create_if_missing=True)
+@app.function(
+ image=image.env({
+ "HF_HOME": f"{CACHE_DIR}/huggingface",
+ "POLICYENGINE_MACRO_PE_DATA_DIR": f"{CACHE_DIR}/policyengine-data",
+ }),
+ cpu=4,
+ memory=8192,
+ # 30 minutes. This function is NOT behind the HTTP proxy, so the 150s
+ # web-endpoint ceiling does not apply to it -- that is the entire point.
+ # The ceiling is why it exists: score_reform over its default five-year
+ # window is two full 372-equation solves plus one PolicyEngine static
+ # costing per year, which does not fit in 150s on this hardware and
+ # cannot be made to.
+ timeout=1800,
+ min_containers=0,
+ scaledown_window=300,
+ max_containers=3,
+ secrets=[modal.Secret.from_name("macromod-hf")],
+ volumes={CACHE_DIR: pe_data_volume},
+)
+def run_tool_job(tool: str, arguments: dict) -> dict:
+ """Worker for start_job: run one allow-listed adapter call to completion."""
+ import os
+
+ if "HUGGING_FACE_TOKEN" not in os.environ and os.environ.get("HF_TOKEN"):
+ os.environ["HUGGING_FACE_TOKEN"] = os.environ["HF_TOKEN"]
+ from policyengine_macro import jobs
+
+ return jobs.run(tool, arguments)
+
+
@app.function(
image=image.env({
"HF_HOME": f"{CACHE_DIR}/huggingface",
@@ -320,9 +351,24 @@ def serve():
if "HUGGING_FACE_TOKEN" not in os.environ and os.environ.get("HF_TOKEN"):
os.environ["HUGGING_FACE_TOKEN"] = os.environ["HF_TOKEN"]
- from policyengine_macro import core
+ from policyengine_macro import core, jobs
from policyengine_macro.mcp_server import mcp
+ # Wire the job tools to Modal. Done here rather than at import time so the
+ # local stdio server and the CLI keep no Modal dependency and report the
+ # absence plainly instead of queueing work nothing will run.
+ def _spawn(tool: str, arguments: dict) -> str:
+ return run_tool_job.spawn(tool, arguments).object_id
+
+ def _poll(job_id: str, wait_seconds: int):
+ call = modal.FunctionCall.from_id(job_id)
+ try:
+ return True, call.get(timeout=wait_seconds)
+ except TimeoutError:
+ return False, None
+
+ jobs.set_backend(_spawn, _poll)
+
# Warm the cheap in-process cache (parses committed results/*.md only —
# NOT a model estimation, which would make cold starts take minutes).
try:
diff --git a/integration/src/policyengine_macro/jobs.py b/integration/src/policyengine_macro/jobs.py
new file mode 100644
index 0000000..68bcc6d
--- /dev/null
+++ b/integration/src/policyengine_macro/jobs.py
@@ -0,0 +1,138 @@
+"""Background jobs for adapter calls that outlive an HTTP request.
+
+Modal enforces a hard **150-second** ceiling on any HTTP request to a web
+endpoint. Past that the proxy abandons the request and returns a 303 pointing
+at a polling URL. That escape hatch does not work here: a 303 tells a
+standards-compliant client to re-issue the request as a GET, and Modal's
+polling URL rejects GET with ``400 modal-http: bad redirect method``. So every
+HTTP client that follows redirects correctly -- curl -L, fetch, httpx with
+follow_redirects -- gets a 400, and every client that does not gets a bare 303.
+There is no client-side fix.
+
+Measured against the deployed server:
+
+ obr_shock ~103s warm -> succeeds; cold -> exceeds 150s and fails
+ score_reform ~111s local, more on Modal -> exceeds 150s, always fails
+
+score_reform cannot be squeezed under the ceiling: the work is two full
+372-equation solves plus one PolicyEngine static costing per year of the
+window. The ceiling is a property of the transport, so the fix is to stop
+doing the work inside the request.
+
+``start_job`` hands the call to a worker with a 30-minute budget and returns
+immediately; ``get_job_result`` polls it, blocking for a bounded interval that
+is itself comfortably inside the 150s ceiling.
+
+This module is deliberately backend-agnostic. The hosted server installs a
+Modal backend at startup (see ``modal_app.serve``); anywhere else -- the local
+stdio server, the CLI, tests -- no backend is installed and the job tools
+report that plainly instead of pretending to queue work that nothing will run.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from policyengine_macro import core
+
+# Adapter calls allowed to run as jobs. An allow-list, not getattr on a
+# caller-supplied name: `tool` arrives from an MCP client, and resolving it
+# straight onto a module would expose every callable in `core`.
+# Names are the ADAPTER function in `core`, which is not always the MCP tool
+# name (the tool `population_reform_impact` is `core.pe_population_impact`,
+# and `dynamic_reform_impact` is `core.dynamic_population_reform_impact`).
+# test_every_allow_listed_tool_resolves_to_a_real_adapter pins that these
+# resolve -- it caught two names that did not.
+JOB_TOOLS: tuple[str, ...] = (
+ "score_reform",
+ "obr_shock",
+ "dynamic_population_reform_impact",
+ "frbus_shock",
+ "hank_shock",
+ "pe_population_impact",
+)
+
+# The longest a get_job_result call may block. The transport ceiling is 150s;
+# this leaves room for request overhead so the poll itself never becomes the
+# thing that times out.
+MAX_WAIT_SECONDS = 120
+
+_spawn: Callable[[str, dict], str] | None = None
+_poll: Callable[[str, int], Any] | None = None
+
+
+def set_backend(spawn, poll) -> None:
+ """Install the job backend. Called once, by the hosted server at startup."""
+ global _spawn, _poll
+ _spawn, _poll = spawn, poll
+
+
+def backend_available() -> bool:
+ return _spawn is not None and _poll is not None
+
+
+class NoBackend(RuntimeError):
+ """Raised where jobs are not available -- i.e. everywhere but the hosted server."""
+
+
+_NO_BACKEND_MESSAGE = (
+ "Background jobs run only on the hosted PolicyEngine Macro server. This "
+ "process has no job backend, so there is nothing to queue the work onto. "
+ "Call the tool directly instead ({tools}) -- running locally there is no "
+ "150-second HTTP ceiling to work around."
+)
+
+
+def resolve(tool: str) -> Callable[..., Any]:
+ """Look up an allow-listed adapter callable by tool name."""
+ if tool not in JOB_TOOLS:
+ raise ValueError(
+ f"{tool!r} cannot be run as a job. Allowed: {', '.join(JOB_TOOLS)}. "
+ "Fast tools should be called directly -- they return well inside "
+ "the transport's 150-second ceiling."
+ )
+ return getattr(core, tool)
+
+
+def run(tool: str, arguments: dict | None) -> dict:
+ """Execute an allow-listed adapter call. This is what the worker runs."""
+ return resolve(tool)(**(arguments or {}))
+
+
+def start(tool: str, arguments: dict | None = None) -> dict:
+ """Queue an adapter call and return a handle without waiting for it."""
+ resolve(tool) # validate before spawning, so a typo fails in milliseconds
+ if not backend_available():
+ raise NoBackend(_NO_BACKEND_MESSAGE.format(tools=", ".join(JOB_TOOLS)))
+ job_id = _spawn(tool, arguments or {})
+ return {
+ "job_id": job_id,
+ "tool": tool,
+ "status": "running",
+ "next_step": (
+ f"Call get_job_result(job_id={job_id!r}). It blocks until the job "
+ f"finishes or up to wait_seconds (max {MAX_WAIT_SECONDS}); if it "
+ "returns status 'running', call it again with the same job_id."
+ ),
+ }
+
+
+def result(job_id: str, wait_seconds: int = 60) -> dict:
+ """Poll a job. Blocks up to wait_seconds, then reports back either way."""
+ if not backend_available():
+ raise NoBackend(_NO_BACKEND_MESSAGE.format(tools=", ".join(JOB_TOOLS)))
+ wait = max(0, min(int(wait_seconds), MAX_WAIT_SECONDS))
+ done, payload = _poll(job_id, wait)
+ if done:
+ return {"job_id": job_id, "status": "done", "result": payload}
+ return {
+ "job_id": job_id,
+ "status": "running",
+ "waited_seconds": wait,
+ "next_step": (
+ "Not finished yet. Call get_job_result again with the same "
+ "job_id; a score_reform over the default five-year window "
+ "typically needs two or three polls."
+ ),
+ }
diff --git a/integration/src/policyengine_macro/mcp_server.py b/integration/src/policyengine_macro/mcp_server.py
index cb99030..533fda2 100644
--- a/integration/src/policyengine_macro/mcp_server.py
+++ b/integration/src/policyengine_macro/mcp_server.py
@@ -14,11 +14,53 @@
from policyengine_macro import core
from policyengine_macro import capabilities
+from policyengine_macro import jobs
from policyengine_macro import reporting
mcp = FastMCP("policyengine-macro")
+@mcp.tool()
+def start_job(
+ tool: Annotated[str, Field(description="Adapter tool to run as a job")],
+ arguments: Annotated[
+ dict | None, Field(description="Arguments for that tool")
+ ] = None,
+) -> dict:
+ """Run a slow tool in the background and get a job id back immediately.
+
+ USE THIS FOR score_reform. The hosted server sits behind a transport that
+ abandons any HTTP request after 150 seconds, and score_reform over its
+ default five-year window takes longer than that, so calling it directly
+ on the hosted server fails every time. obr_shock, frbus_shock, hank_shock,
+ dynamic_reform_impact and population_reform_impact are also close enough
+ to the ceiling to fail on a cold server.
+
+ Returns a job id straight away. Poll it with get_job_result. The job
+ itself has a 30-minute budget, so the 150-second ceiling stops applying.
+
+ Running locally (stdio server or the pe-macro CLI) there is no such
+ ceiling and no job backend: call the tool directly instead.
+ """
+ return jobs.start(tool, arguments)
+
+
+@mcp.tool()
+def get_job_result(
+ job_id: Annotated[str, Field(description="Job id from start_job")],
+ wait_seconds: Annotated[
+ int, Field(description="Seconds to block waiting, max 120")
+ ] = 60,
+) -> dict:
+ """Fetch a started job's result, waiting up to wait_seconds for it.
+
+ Returns status 'done' with the result, or status 'running' -- in which
+ case call again with the same job_id. A score_reform over the default
+ five-year window typically needs two or three polls.
+ """
+ return jobs.result(job_id, wait_seconds)
+
+
@mcp.tool()
def list_model_capabilities() -> list[dict]:
"""List supported questions, outputs, access, runtime, evidence status,
diff --git a/integration/tests/test_jobs.py b/integration/tests/test_jobs.py
new file mode 100644
index 0000000..80dd0eb
--- /dev/null
+++ b/integration/tests/test_jobs.py
@@ -0,0 +1,171 @@
+"""The background-job escape hatch from the transport's 150-second ceiling.
+
+Modal abandons any HTTP request to a web endpoint after 150 seconds. Its
+documented escape hatch -- a 303 to a polling URL -- does not work for an MCP
+server: 303 means "re-issue as GET", and Modal's polling URL answers GET with
+`400 modal-http: bad redirect method`. Measured against the live deployment,
+every correctly-behaving client gets a 400 and every other client gets a bare
+303. So score_reform, which needs longer than 150s over its default five-year
+window, could not be called on the hosted server at all.
+
+These cover the parts that are testable without Modal: the allow-list, the
+handle contract, and what happens where no backend is installed (which is
+everywhere except the hosted server -- the local stdio server, the CLI, and
+this test process). The Modal round-trip itself is covered by the post-deploy
+smoke test in test_remote_mcp.py.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from policyengine_macro import jobs
+
+
+@pytest.fixture(autouse=True)
+def _no_backend():
+ """Every test here runs with no backend, and none may leak to the next."""
+ before = (jobs._spawn, jobs._poll)
+ jobs.set_backend(None, None)
+ yield
+ jobs.set_backend(*before)
+
+
+def test_score_reform_is_runnable_as_a_job():
+ """The tool the ceiling actually blocks must be on the allow-list."""
+ assert "score_reform" in jobs.JOB_TOOLS
+ assert jobs.resolve("score_reform").__name__ == "score_reform"
+
+
+def test_every_allow_listed_tool_resolves_to_a_real_adapter():
+ """No entry may name a callable that does not exist."""
+ for tool in jobs.JOB_TOOLS:
+ assert callable(jobs.resolve(tool)), tool
+
+
+def test_unknown_tool_is_refused_with_the_allowed_set():
+ """`tool` comes from an MCP client, so it is an allow-list, not getattr.
+
+ Resolving a caller-supplied name straight onto the module would expose
+ every callable in `core` -- including private helpers -- to anyone who
+ can reach the server.
+ """
+ with pytest.raises(ValueError) as excinfo:
+ jobs.resolve("__import__")
+ message = str(excinfo.value)
+ assert "cannot be run as a job" in message
+ assert "score_reform" in message, "the refusal does not say what IS allowed"
+
+
+def test_fast_tools_are_not_job_runnable():
+ """Only the slow tools. A job handle for a sub-second call is pure overhead."""
+ for tool in ("list_model_capabilities", "calculate_household", "forecast_uk"):
+ with pytest.raises(ValueError):
+ jobs.resolve(tool)
+
+
+def test_without_a_backend_start_says_so_and_says_what_to_do_instead():
+ """Locally there is no ceiling and no backend: say that, do not pretend.
+
+ Silently accepting the call and handing back a job id that nothing will
+ ever run would be the worst outcome -- the caller would poll forever.
+ """
+ with pytest.raises(jobs.NoBackend) as excinfo:
+ jobs.start("score_reform", {"country": "uk"})
+ message = str(excinfo.value)
+ assert "hosted" in message
+ assert "Call the tool directly instead" in message
+
+
+def test_without_a_backend_result_also_refuses():
+ with pytest.raises(jobs.NoBackend):
+ jobs.result("fc-whatever")
+
+
+def test_a_bad_tool_name_fails_before_anything_is_spawned():
+ """Validation precedes the spawn, so a typo costs milliseconds, not a container."""
+ spawned = []
+ jobs.set_backend(lambda t, a: spawned.append((t, a)) or "fc-1", lambda j, w: (True, {}))
+ with pytest.raises(ValueError):
+ jobs.start("not_a_tool", {})
+ assert spawned == [], "a rejected tool still reached the backend"
+
+
+def test_start_returns_a_handle_that_says_how_to_collect_it():
+ calls = []
+ jobs.set_backend(lambda t, a: calls.append((t, a)) or "fc-abc", lambda j, w: (True, {}))
+ out = jobs.start("score_reform", {"country": "uk", "model": "obr"})
+ assert out["job_id"] == "fc-abc"
+ assert out["tool"] == "score_reform"
+ assert out["status"] == "running"
+ # The handle has to carry the next step: an agent that gets a job id and
+ # no instruction has no way to know get_job_result exists.
+ assert "get_job_result" in out["next_step"]
+ assert calls == [("score_reform", {"country": "uk", "model": "obr"})]
+
+
+def test_start_passes_an_empty_dict_rather_than_none():
+ """The worker calls fn(**arguments); None would raise inside the container."""
+ calls = []
+ jobs.set_backend(lambda t, a: calls.append(a) or "fc-1", lambda j, w: (True, {}))
+ jobs.start("obr_shock")
+ assert calls == [{}]
+
+
+def test_finished_job_returns_the_result():
+ jobs.set_backend(lambda t, a: "fc-1", lambda j, w: (True, {"gdp": -1.0}))
+ out = jobs.result("fc-1")
+ assert out["status"] == "done"
+ assert out["result"] == {"gdp": -1.0}
+
+
+def test_unfinished_job_tells_the_caller_to_poll_again():
+ jobs.set_backend(lambda t, a: "fc-1", lambda j, w: (False, None))
+ out = jobs.result("fc-1", wait_seconds=30)
+ assert out["status"] == "running"
+ assert out["job_id"] == "fc-1"
+ assert out["waited_seconds"] == 30
+ assert "again" in out["next_step"]
+
+
+def test_wait_is_capped_below_the_transport_ceiling():
+ """A poll that outlived the 150s ceiling would be the bug it works around."""
+ seen = []
+ jobs.set_backend(lambda t, a: "fc-1", lambda j, w: (seen.append(w), (False, None))[1])
+ jobs.result("fc-1", wait_seconds=10_000)
+ assert seen == [jobs.MAX_WAIT_SECONDS]
+ assert jobs.MAX_WAIT_SECONDS < 150, "the cap must sit inside the ceiling"
+
+
+def test_negative_wait_is_clamped_not_passed_through():
+ seen = []
+ jobs.set_backend(lambda t, a: "fc-1", lambda j, w: (seen.append(w), (False, None))[1])
+ jobs.result("fc-1", wait_seconds=-5)
+ assert seen == [0]
+
+
+def test_job_tools_are_on_the_mcp_surface():
+ """They are useless if a client cannot see them."""
+ import asyncio
+
+ from policyengine_macro.mcp_server import mcp
+
+ names = {t.name for t in asyncio.run(mcp.list_tools())}
+ assert {"start_job", "get_job_result"} <= names
+
+
+def test_start_job_tool_description_names_score_reform():
+ """An agent hitting the ceiling has to be able to find the way round it.
+
+ The description is the only place a client learns that score_reform needs
+ the job path on the hosted server, so this pins the pointer rather than
+ the prose.
+ """
+ import asyncio
+
+ from policyengine_macro.mcp_server import mcp
+
+ tools = {t.name: t for t in asyncio.run(mcp.list_tools())}
+ description = tools["start_job"].description
+ assert "score_reform" in description
+ assert "150" in description
diff --git a/integration/tests/test_remote_mcp.py b/integration/tests/test_remote_mcp.py
index 545bec5..a2e809f 100644
--- a/integration/tests/test_remote_mcp.py
+++ b/integration/tests/test_remote_mcp.py
@@ -376,3 +376,76 @@ async def test_calculate_household_uk_50k():
@pytest.fixture
def anyio_backend():
return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_score_reform_default_window_works_through_a_job():
+ """The path the ceiling used to make impossible.
+
+ Modal abandons any HTTP request after 150 seconds. score_reform over its
+ DEFAULT five-year window needs longer than that, so calling it directly on
+ the hosted server fails every time -- measured: HTTP 303 at 150.3s, and
+ following that redirect gives `400 modal-http: bad redirect method`
+ because a 303 turns the POST into a GET.
+
+ Note the existing bridge test uses years=1 to stay under the ceiling,
+ which is exactly why the ceiling survived so long without CI noticing:
+ nothing in the smoke suite exercised the default a real caller gets.
+ This does.
+ """
+ import asyncio
+
+ started = await _call(
+ "start_job",
+ {
+ "tool": "score_reform",
+ "arguments": {
+ "country": "uk",
+ "reform": {"gov.hmrc.income_tax.rates.uk[0].rate": 0.21},
+ "model": "obr",
+ },
+ },
+ )
+ assert started["status"] == "running", started
+ job_id = started["job_id"]
+ assert job_id, started
+ # The handle must tell an agent how to collect it.
+ assert "get_job_result" in started["next_step"]
+
+ # Poll. Each call blocks server-side well inside the 150s ceiling, so the
+ # loop is what makes an arbitrarily long job reachable over this transport.
+ # Bounded so a broken job path fails the deploy quickly rather than
+ # sitting on the runner: deploy (~4 min) + this must stay under the
+ # workflow's 25-minute cap.
+ deadline = 10 * 60
+ waited = 0
+ out = None
+ while waited < deadline:
+ out = await asyncio.wait_for(
+ _call("get_job_result", {"job_id": job_id, "wait_seconds": 120}),
+ timeout=180,
+ )
+ if out["status"] == "done":
+ break
+ assert out["status"] == "running", out
+ waited += out["waited_seconds"] or 1
+ assert out is not None and out["status"] == "done", (
+ f"score_reform did not finish within {deadline}s: {out}"
+ )
+
+ result = out["result"]
+ # Same invariants as the direct bridge test, over the full default window.
+ assert result["bridge_variable"] == "HHDI_ADDFACTOR"
+ assert len(result["annual_costings_bn"]) == 5, "not the default 5-year window"
+ assert all(c["budgetary_impact_bn"] > 0 for c in result["annual_costings_bn"])
+ assert result["cumulative_delta_gdp_bn_over_shock_periods"] < 0
+ assert result["score"]["caveats"]
+
+
+@pytest.mark.anyio
+async def test_job_tools_refuse_a_tool_that_is_not_allow_listed():
+ """`tool` is caller-supplied, so the server must not getattr it onto core."""
+ text = await _call_expecting_error(
+ "start_job", {"tool": "__import__", "arguments": {}}
+ )
+ assert "cannot be run as a job" in text
diff --git a/integration/tests/tool_surface.py b/integration/tests/tool_surface.py
index 86d7792..6c6c672 100644
--- a/integration/tests/tool_surface.py
+++ b/integration/tests/tool_surface.py
@@ -21,6 +21,7 @@
"define_scenario_incidence",
"dynamic_reform_impact",
"forecast_uk",
+ "start_job",
"format_score_report",
"frbus_list_variables",
"frbus_shock",
@@ -36,6 +37,7 @@
"list_reform_parameters",
"list_reform_variables",
"model_summary",
+ "get_job_result",
"get_model_status",
"obr_shock",
"population_reform_impact",
@@ -44,7 +46,7 @@
}
)
-GOLDEN_TOOL_COUNT = 26
+GOLDEN_TOOL_COUNT = 28
assert len(GOLDEN_TOOLS) == GOLDEN_TOOL_COUNT