From be985261a4b87157bf041e683b2b19f599c3512b Mon Sep 17 00:00:00 2001 From: Thang Pham Date: Sat, 5 Sep 2026 07:26:30 -0500 Subject: [PATCH] fix: preserve worker interruptions and enforce scientific acceptance --- .github/workflows/ci.yml | 4 +- docs/unified-api.md | 12 +++ src/matkit/api/bundles.py | 22 ++++ src/matkit/api/models.py | 49 +++++++++ src/matkit/api/runtime.py | 46 ++++++--- src/matkit/mcp.py | 39 ++----- tests/fixtures/post_commit_worker.py | 24 +++++ tests/test_api_contracts.py | 122 +++++++++++++++++++++- tests/test_api_worker_outcomes.py | 145 +++++++++++++++++++++++++++ tests/test_mcp_api.py | 62 ++++++++++++ 10 files changed, 476 insertions(+), 49 deletions(-) create mode 100644 tests/fixtures/post_commit_worker.py create mode 100644 tests/test_api_worker_outcomes.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6aa7728..fe1bae4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,8 +26,8 @@ jobs: ruff format --check tests/test_mlip*.py tests/test_cli.py examples/mlip_gpu.py alcf/polaris/mlip/smoke.py - name: Check unified API tests and recipes run: | - ruff check tests/test_api*.py tests/test_foundations.py tests/test_operation_cli.py tests/test_mcp_api.py tests/test_packaged_resources.py tests/test_unified_smoke.py tests/fixtures/fake_engine.py examples/unified_smoke.py - ruff format --check tests/test_api*.py tests/test_foundations.py tests/test_operation_cli.py tests/test_mcp_api.py tests/test_packaged_resources.py tests/test_unified_smoke.py tests/fixtures/fake_engine.py examples/unified_smoke.py + ruff check tests/test_api*.py tests/test_foundations.py tests/test_operation_cli.py tests/test_mcp_api.py tests/test_packaged_resources.py tests/test_unified_smoke.py tests/fixtures/*.py examples/unified_smoke.py + ruff format --check tests/test_api*.py tests/test_foundations.py tests/test_operation_cli.py tests/test_mcp_api.py tests/test_packaged_resources.py tests/test_unified_smoke.py tests/fixtures/*.py examples/unified_smoke.py test: runs-on: ubuntu-latest diff --git a/docs/unified-api.md b/docs/unified-api.md index 3b6aff7..a972650 100644 --- a/docs/unified-api.md +++ b/docs/unified-api.md @@ -193,6 +193,12 @@ An unconverged relaxation can have valid numerical results and `state=completed` while `accepted` is false. Unknown sampling quality does not become a claim of equilibrium. Failure information is separate from adsorption uncertainty. +Completed evaluation records must contain every requested property, including +when read back from JSON. Relaxation acceptance requires both convergence and +an explicit, required, passed `force_convergence` check. Missing checks leave +otherwise valid numerical records unaccepted; contradictory convergence claims +and forces exceeding the requested threshold are rejected during validation. + Energy uses `potential_energy` in eV with a model-specific reference; forces use eV/angstrom; stress uses the ASE Cartesian convention in eV/angstrom³. The shared interface does not make energies from different methods comparable. @@ -229,6 +235,12 @@ can leave a running record or stale lock; automatic resume and restart are not implemented. Copy a completed result for inspection, and prepare a fresh bundle for another execution. Engine-specific restart requires additional future work. +Supervised Python/CLI and MCP execution record timeouts, cancellation, teardown +failures, and unexpected exit codes even after a numerical result was committed. +In that case `result.json` retains the scientific outcome, while `run.json` and +inspection report an orchestration interruption and `accepted=false`. Worker +log inventories are refreshed without replacing the committed scientific state. + ## MCP ```bash diff --git a/src/matkit/api/bundles.py b/src/matkit/api/bundles.py index e0510e4..d35832f 100644 --- a/src/matkit/api/bundles.py +++ b/src/matkit/api/bundles.py @@ -151,6 +151,28 @@ def commit_result(root: Path, result: RunResult) -> RunResult: return result +def refresh_artifacts(root: Path, record: RunResult) -> RunResult: + """Update closed worker logs without replacing a committed outcome. + + The manifest can report interruption after the scientific result was + committed. Keep those states distinct while refreshing both inventories. + """ + artifacts = collect_artifacts(root) + record = RunResult.model_validate( + record.model_copy(update={"artifacts": artifacts}).model_dump() + ) + result_path = root / "result.json" + if result_path.exists(): + committed = RunResult.model_validate_json(result_path.read_text()) + if committed.run_id != record.run_id: + raise ValueError("Committed result does not match run manifest") + atomic_json( + result_path, committed.model_copy(update={"artifacts": artifacts}) + ) + atomic_json(root / "run.json", record) + return record + + def environment_versions() -> dict: versions = {} for package in ( diff --git a/src/matkit/api/models.py b/src/matkit/api/models.py index d147a38..0bcf465 100644 --- a/src/matkit/api/models.py +++ b/src/matkit/api/models.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import math from typing import Annotated, Literal, Union from pydantic import ( @@ -353,10 +354,25 @@ def accepted(self) -> bool: or check.status in {"passed", "not_applicable"} for check in self.checks ) + and ( + self.operation != "relax" + or ( + isinstance(self.payload, EvaluationPayload) + and self.payload.converged is True + and any( + check.name == "force_convergence" + and check.required + and check.status == "passed" + for check in self.checks + ) + ) + ) ) @model_validator(mode="after") def consistent_outcome(self): + if len({check.name for check in self.checks}) != len(self.checks): + raise ValueError("Scientific check names must be unique") if self.state == "completed" and ( self.payload is None or self.numerical_validity != "valid" @@ -375,6 +391,39 @@ def consistent_outcome(self): ) if self.payload.kind != kind: raise ValueError("Payload kind does not match operation") + if self.state == "completed" and isinstance( + self.payload, EvaluationPayload + ): + request = parse_request(self.requested) + if request.operation != self.operation: + raise ValueError("Requested operation does not match result") + required = ( + request.properties + if isinstance(request, EvaluateRequest) + else ["potential_energy", "forces"] + ) + for name in required: + value = getattr(self.payload, name) + if value is None or (name == "forces" and not value): + raise ValueError(f"Requested property {name} is missing") + if isinstance(request, RelaxRequest): + if self.payload.converged is None: + raise ValueError("Relaxation must report convergence") + expected = "passed" if self.payload.converged else "failed" + for check in self.checks: + if check.name == "force_convergence" and ( + not check.required or check.status != expected + ): + raise ValueError( + "Contradictory force convergence check" + ) + max_force = max( + math.hypot(*force) for force in self.payload.forces + ) + if self.payload.converged and ( + max_force > request.fmax * (1 + 1e-6) + 1e-12 + ): + raise ValueError("Converged forces exceed requested fmax") return self diff --git a/src/matkit/api/runtime.py b/src/matkit/api/runtime.py index f142a40..c41ecca 100644 --- a/src/matkit/api/runtime.py +++ b/src/matkit/api/runtime.py @@ -18,6 +18,7 @@ environment_versions, inspect_run, prepare, + refresh_artifacts, staged_request, verify_inputs, ) @@ -174,6 +175,25 @@ def _worker_command(root, execution, batch): return command +def _interrupt_run(root, exc): + """Record interruption without replacing committed science.""" + record = inspect_run(root) + if not (root / "result.json").exists(): + return _fail(root, record, exc, "worker", interrupted=True) + interrupted = record.model_copy( + update={ + "state": "interrupted", + "failure": Failure( + code=type(exc).__name__, + stage="orchestration", + message=str(exc) or type(exc).__name__, + ), + } + ) + atomic_json(root / "run.json", interrupted) + return interrupted + + def _supervise_claimed(root, execution, batch=False): atomic_json( root / "execution.json", @@ -201,22 +221,16 @@ def _supervise_claimed(root, execution, batch=False): adapters.stop_process(process) if batch: _interrupt_batch(root, exc) - elif not (root / "result.json").exists(): - _fail(root, inspect_run(root), exc, "worker", interrupted=True) + else: + result = refresh_artifacts(root, _interrupt_run(root, exc)) if isinstance(exc, subprocess.TimeoutExpired): if batch: return read_batch(root) - result = inspect_run(root) - return commit_result( - root, - result.model_copy( - update={"artifacts": collect_artifacts(root)} - ), - ) + return result raise if batch: result = read_batch(root) - if result.state == "running": + if result.state == "running" or code != (0 if result.accepted else 1): _interrupt_batch( root, RuntimeError(f"Worker exited with code {code}") ) @@ -233,13 +247,13 @@ def _supervise_claimed(root, execution, batch=False): interrupted=True, ) if code != (0 if result.accepted else 1): - raise RuntimeError( - f"Worker exited with code {code}; " - f"committed results remain in {root}" + result = _interrupt_run( + root, + RuntimeError( + f"Worker exited with code {code}; see worker.stderr.log" + ), ) - return commit_result( - root, result.model_copy(update={"artifacts": collect_artifacts(root)}) - ) + return refresh_artifacts(root, result) def execute( diff --git a/src/matkit/mcp.py b/src/matkit/mcp.py index 35dabd5..00c5eb0 100644 --- a/src/matkit/mcp.py +++ b/src/matkit/mcp.py @@ -26,12 +26,11 @@ artifact, atomic_json, claim, - collect_artifacts, - commit_result, contained_path, + refresh_artifacts, ) from matkit.api.models import Artifact, Failure, Model, ScientificCheck -from matkit.api.runtime import _fail, _worker_command +from matkit.api.runtime import _interrupt_run, _worker_command from matkit.api.structures import sha256 from matkit.operation_cli import resolve_request_paths @@ -126,16 +125,15 @@ async def _execute_bounded(root, execution): with anyio.fail_after(execution.timeout_s): code = await process.wait() record = inspect_run(root) - if record.state in {"prepared", "running"}: - _fail( + if record.state in {"prepared", "running"} or code != ( + 0 if record.accepted else 1 + ): + _interrupt_run( root, - record, RuntimeError( f"Worker exited with code {code}; " "see worker.stderr.log" ), - "worker", - interrupted=True, ) except BaseException as exc: with anyio.CancelScope(shield=True): @@ -152,14 +150,7 @@ async def _execute_bounded(root, execution): else: process.kill() await process.wait() - if not (root / "result.json").exists(): - _fail( - root, - inspect_run(root), - exc, - "worker", - interrupted=True, - ) + _interrupt_run(root, exc) if not isinstance(exc, (TimeoutError, OSError)): raise finally: @@ -173,22 +164,10 @@ async def _execute_bounded(root, execution): await process.aclose() record = inspect_run(root) if record.state not in {"prepared", "running"}: - commit_result( - root, - record.model_copy( - update={ - "artifacts": collect_artifacts(root) - } - ), - ) + refresh_artifacts(root, record) record = inspect_run(root) if record.state not in {"prepared", "running"}: - commit_result( - root, - record.model_copy( - update={"artifacts": collect_artifacts(root)} - ), - ) + refresh_artifacts(root, record) return _summary(root) diff --git a/tests/fixtures/post_commit_worker.py b/tests/fixtures/post_commit_worker.py new file mode 100644 index 0000000..ce689e1 --- /dev/null +++ b/tests/fixtures/post_commit_worker.py @@ -0,0 +1,24 @@ +"""Exercise real supervision after a synthetic calculation has committed.""" + +from pathlib import Path +import sys +import time + +from matkit.api import ExecutionConfig +from matkit.api.runtime import _execute_batch_claimed, _execute_claimed + +root = Path(sys.argv[1]) +behavior = sys.argv[2] +config = ExecutionConfig.model_validate_json( + (root / "execution.json").read_text() +) +execute = _execute_batch_claimed if "--batch" in sys.argv else _execute_claimed +result = execute(root, config) +assert result.accepted +(root / "worker.finished").write_text("scientific result committed") +if behavior == "hang": + time.sleep(60) +elif behavior == "teardown_error": + raise RuntimeError("fixture teardown failed after commit") +elif behavior == "bad_exit": + raise SystemExit(7) diff --git a/tests/test_api_contracts.py b/tests/test_api_contracts.py index 6f0c6ad..6ee7319 100644 --- a/tests/test_api_contracts.py +++ b/tests/test_api_contracts.py @@ -22,7 +22,7 @@ parse_request, prepare, ) -from matkit.api.models import REQUEST_ADAPTER +from matkit.api.models import REQUEST_ADAPTER, EvaluationPayload from matkit.api.structures import final_structure, load_structure, to_atoms @@ -148,3 +148,123 @@ def test_preparation_is_engine_independent(sample_cif, tmp_path): assert record.state == "prepared" assert any(a.path == "inputs/radii.rad" for a in record.artifacts) assert all(not Path(a.path).is_absolute() for a in record.artifacts) + + +def _relaxation_record(converged, checks): + request = RelaxRequest( + structure=StructureRef(path="input.cif"), + method=MLIPMethod(checkpoint="fixture"), + ) + return { + "run_id": "fixture", + "operation": "relax", + "state": "completed", + "numerical_validity": "valid", + "requested": request.model_dump(), + "payload": EvaluationPayload( + potential_energy=-1, + forces=[[0, 0, 0]], + converged=converged, + ).model_dump(), + "checks": checks, + } + + +@pytest.mark.parametrize("converged", [True, False]) +def test_missing_convergence_check_never_accepts_imported_result(converged): + record = RunResult.model_validate_json( + json.dumps(_relaxation_record(converged, [])) + ) + assert record.numerical_validity == "valid" + assert not record.accepted + + +@pytest.mark.parametrize( + "converged,status", [(False, "passed"), (True, "failed")] +) +def test_contradictory_imported_convergence_rejected(converged, status): + values = _relaxation_record( + converged, + [ + { + "name": "force_convergence", + "required": True, + "status": status, + } + ], + ) + with pytest.raises(ValueError, match="Contradictory"): + RunResult.model_validate_json(json.dumps(values)) + + +@pytest.mark.parametrize("converged", [True, False]) +def test_valid_convergence_round_trip(converged): + values = _relaxation_record( + converged, + [ + { + "name": "force_convergence", + "required": True, + "status": "passed" if converged else "failed", + } + ], + ) + result = RunResult.model_validate_json(json.dumps(values)) + assert result.accepted is converged + assert RunResult.model_validate_json(result.model_dump_json()) == result + + +def test_converged_result_requires_forces_below_requested_threshold(): + values = _relaxation_record( + True, + [ + { + "name": "force_convergence", + "required": True, + "status": "passed", + } + ], + ) + values["payload"]["forces"] = [[1, 0, 0]] + with pytest.raises(ValueError, match="exceed requested fmax"): + RunResult.model_validate_json(json.dumps(values)) + + +@pytest.mark.parametrize( + "properties", + [ + ["potential_energy"], + ["forces"], + ["stress"], + ["potential_energy", "forces", "stress"], + ], +) +def test_completed_evaluation_requires_requested_properties(properties): + request = EvaluateRequest( + structure=StructureRef(path="input.cif"), + method=MLIPMethod(checkpoint="fixture"), + properties=properties, + ) + values = { + "run_id": "fixture", + "operation": "evaluate", + "state": "completed", + "numerical_validity": "valid", + "requested": request.model_dump(), + "payload": { + "kind": "evaluation", + "potential_energy": -1, + "forces": [[0, 0, 0]], + "stress": [[0, 0, 0]] * 3, + }, + } + result = RunResult.model_validate_json(json.dumps(values)) + assert result.accepted + for property_name in properties: + incomplete = json.loads(json.dumps(values)) + incomplete["payload"][property_name] = None + with pytest.raises(ValueError, match=f"property {property_name}"): + RunResult.model_validate_json(json.dumps(incomplete)) + for property_name in set(values["payload"]) - set(properties) - {"kind"}: + values["payload"][property_name] = None + assert RunResult.model_validate_json(json.dumps(values)).accepted diff --git a/tests/test_api_worker_outcomes.py b/tests/test_api_worker_outcomes.py new file mode 100644 index 0000000..398d551 --- /dev/null +++ b/tests/test_api_worker_outcomes.py @@ -0,0 +1,145 @@ +"""Committed science and worker failure must remain independently visible.""" + +import json +from pathlib import Path +import sys +import time + +from click.testing import CliRunner +import pytest + +from matkit.api import ( + PoreRequest, + RunResult, + StructureRef, + inspect_run, + prepare, + run, + run_batch, +) +from matkit.api import runtime +from matkit.api.structures import sha256 +from matkit.cli import main +from tests.test_api_runtime import fake_execution + + +def worker_command(behavior): + fixture = Path(__file__).parent / "fixtures" / "post_commit_worker.py" + + def command(root, execution, batch): + args = [sys.executable, str(fixture), str(root), behavior] + return [*args, "--batch"] if batch else args + + return command + + +def assert_preserved(root): + committed = RunResult.model_validate_json( + (root / "result.json").read_text() + ) + inspected = inspect_run(root) + assert committed.accepted + assert inspected.state == "interrupted" + assert not inspected.accepted + assert inspected.failure.stage == "orchestration" + assert inspected.numerical_validity == "valid" + assert inspected.payload == committed.payload + assert (root / "worker.finished").exists() + assert not (root / ".matkit.lock").exists() + assert all(sha256(root / a.path) == a.sha256 for a in inspected.artifacts) + return inspected + + +@pytest.mark.parametrize("behavior", ["hang", "bad_exit", "teardown_error"]) +def test_post_commit_worker_failures( + behavior, sample_cif, tmp_path, monkeypatch +): + monkeypatch.setattr(runtime, "_worker_command", worker_command(behavior)) + root = tmp_path / "run" + result = run( + PoreRequest(structure=StructureRef(path=sample_cif)), + output_dir=root, + execution=fake_execution("zeopp", mode="subprocess", timeout_s=3), + ) + assert result == assert_preserved(root) + if behavior == "hang": + assert result.failure.code == "TimeoutExpired" + + +def test_cli_reports_post_commit_failure(sample_cif, tmp_path, monkeypatch): + monkeypatch.setattr(runtime, "_worker_command", worker_command("bad_exit")) + root = tmp_path / "run" + prepare( + PoreRequest(structure=StructureRef(path=sample_cif)), output_dir=root + ) + profile = tmp_path / "execution.json" + profile.write_text(fake_execution("zeopp").model_dump_json()) + runner = CliRunner() + response = runner.invoke( + main, + ["execute", str(root), "--execution", str(profile)], + ) + assert response.exit_code == 1, response.output + assert json.loads(response.stdout)["state"] == "interrupted" + assert_preserved(root) + inspected = runner.invoke(main, ["inspect", str(root)]) + assert inspected.exit_code == 0, inspected.output + assert json.loads(inspected.stdout)["state"] == "interrupted" + + +def test_keyboard_interrupt_preserves_committed_result( + sample_cif, tmp_path, monkeypatch +): + monkeypatch.setattr(runtime, "_worker_command", worker_command("hang")) + root = tmp_path / "run" + original_popen = runtime.subprocess.Popen + + def launch(*args, **kwargs): + process = original_popen(*args, **kwargs) + original_wait = process.wait + interrupted = False + + def wait(timeout=None): + nonlocal interrupted + if not interrupted: + deadline = time.monotonic() + 5 + while not (root / "worker.finished").exists(): + if time.monotonic() >= deadline: + raise AssertionError("worker did not commit") + time.sleep(0.025) + interrupted = True + raise KeyboardInterrupt("fixture cancellation") + return original_wait(timeout=timeout) + + process.wait = wait + return process + + monkeypatch.setattr(runtime.subprocess, "Popen", launch) + with pytest.raises(KeyboardInterrupt, match="fixture cancellation"): + run( + PoreRequest(structure=StructureRef(path=sample_cif)), + output_dir=root, + execution=fake_execution("zeopp", mode="subprocess"), + ) + assert_preserved(root) + + +@pytest.mark.parametrize("behavior", ["hang", "bad_exit"]) +def test_batch_retains_completed_items_after_worker_failure( + behavior, sample_cif, tmp_path, monkeypatch +): + monkeypatch.setattr(runtime, "_worker_command", worker_command(behavior)) + request = PoreRequest(structure=StructureRef(path=sample_cif)) + root = tmp_path / "batch" + result = run_batch( + [request, request], + output_dir=root, + execution=fake_execution("zeopp", mode="subprocess", timeout_s=3), + ) + assert result.state == "interrupted" + assert not result.accepted + assert result.failure is not None + assert all(item["accepted"] for item in result.items) + assert all( + inspect_run(root / item["bundle"]).accepted for item in result.items + ) diff --git a/tests/test_mcp_api.py b/tests/test_mcp_api.py index da08267..494d6af 100644 --- a/tests/test_mcp_api.py +++ b/tests/test_mcp_api.py @@ -163,3 +163,65 @@ async def check(): pytest.fail("cancellation did not persist the interrupted run") asyncio.run(check()) + + +@pytest.mark.parametrize( + "behavior", ["hang", "bad_exit", "teardown_error", "cancel"] +) +def test_mcp_preserves_result_and_reports_post_commit_failure( + behavior, sample_cif, tmp_path, monkeypatch +): + from matkit import mcp + from tests.test_api_worker_outcomes import assert_preserved, worker_command + + monkeypatch.setattr( + mcp, + "_worker_command", + worker_command("hang" if behavior == "cancel" else behavior), + ) + root = tmp_path / "runs" + server = create_server( + run_root=root, + input_roots=[Path(sample_cif).parent], + profiles={"default": profile()}, + timeout_s=30 if behavior == "cancel" else 3, + ) + + async def check(): + async with Client(server) as client: + task = asyncio.create_task( + client.call_tool( + "matkit_pores", + {"request": {"structure": {"path": sample_cif}}}, + ) + ) + if behavior == "cancel": + for _ in range(200): + if list(root.glob("*/worker.finished")): + break + await asyncio.sleep(0.025) + else: + pytest.fail("worker did not commit its result") + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + for _ in range(200): + run_root = next(root.glob("*/worker.finished")).parent + if ( + not (run_root / ".matkit.lock").exists() + and inspect_run(run_root).state == "interrupted" + ): + break + await asyncio.sleep(0.025) + else: + pytest.fail("cancellation did not record interruption") + else: + response = await task + assert not response.is_error, response + data = response.structured_content + assert data["state"] == "interrupted" + assert not data["accepted"] + run_root = next(root.glob("*/worker.finished")).parent + assert_preserved(run_root) + + asyncio.run(check())