decomposer: generate files for Maintain only the latest check result for each monitored target in concurrency-safe memory and expose it through the specified FastAPI current-status endpoint.

src/current_status/transform.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import copy
+import json
+import sys
+from typing import Any
+
+_REQUIRED = (
+    "service_specification",
+    "target_management_implementation",
+    "endpoint_check_implementation",
+)
+
+
+def transform(value: dict[str, Any]) -> dict[str, Any]:
+    """Preserve all prior contract objects and append this step's implementation."""
+    if not isinstance(value, dict):
+        raise TypeError("input must be an object")
+    for key in _REQUIRED:
+        if not isinstance(value.get(key), dict):
+            raise ValueError(f"{key} must be an object")
+
+    output = copy.deepcopy(value)
+    output["current_status_implementation"] = {
+        "language": "python",
+        "framework": "FastAPI",
+        "package": "current_status",
+        "coordinator": "current_status.CurrentStatusCoordinator",
+        "router_factory": "current_status.build_current_status_router",
+        "endpoint": "GET /targets/{target_id}/status",
+        "storage": "asyncio.Lock guarded, one status dictionary entry per target",
+        "no_check_yet": {"http_status": 200, "state": "not_checked", "result": None},
+        "not_found": {"http_status": 404, "code": "target_not_found"},
+        "lifecycle_hooks": {
+            "create": "await coordinator.register_target(target_id)",
+            "delete": "await coordinator.delete_target(target_id)",
+            "check": "await coordinator.run_check(target_id, target, checker)",
+        },
+        "race_guards": [
+            "target incarnation rejects results from deleted/recreated targets",
+            "monotonic start sequence rejects older overlapping checks",
+        ],
+        "tests": [
+            "tests/test_api.py",
+            "tests/test_concurrency.py",
+            "tests/test_transform.py",
+        ],
+    }
+    return output
+
+
+def main() -> None:
+    try:
+        result = transform(json.load(sys.stdin))
+        json.dump(result, sys.stdout, separators=(",", ":"))
+        sys.stdout.write("\n")
+    except (TypeError, ValueError, json.JSONDecodeError) as exc:
+        print(str(exc), file=sys.stderr)
+        raise SystemExit(2) from exc
+
+
+if __name__ == "__main__":
+    main()