(runtime_kind: str, **fields: Any)
| 36 | |
| 37 | |
| 38 | def _update_status(runtime_kind: str, **fields: Any) -> Dict[str, Any]: |
| 39 | kind = _normalize_runtime_kind(runtime_kind) |
| 40 | with _STATUS_LOCK: |
| 41 | status = _RUNTIME_STATUS[kind] |
| 42 | status.update(fields) |
| 43 | status["updated_at"] = _now_iso() |
| 44 | status["sequence"] = int(status.get("sequence", 0) or 0) + 1 |
| 45 | if status.get("state") == "ready": |
| 46 | status["last_completed_at"] = status["updated_at"] |
| 47 | return deepcopy(status) |
| 48 | |
| 49 | |
| 50 | def start_runtime_prepare(runtime_kind: str, message: str) -> Dict[str, Any]: |
no test coverage detected