fix: preserve worker failures and enforce scientific acceptance

.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

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

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 (

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
 
 

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(

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)
 
 

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)