"""Minimal HTTP surface for the independent DataOps Runner.""" import hashlib import json import re from flask import Flask, jsonify, request from app.runner.auth import TaskTokenExpired, TaskTokenInvalid from app.runner.nodes import NodeExecutionError def create_runner_app(*, verifier, ledger, registry): app = Flask("dataops-runner") app.config["MAX_CONTENT_LENGTH"] = 2 * 1024 * 1024 @app.get("/health") def health(): return jsonify({"status": "ok"}) def finish_safely(jti, **outcome): try: ledger.finish(jti, **outcome) return True except Exception: app.logger.error("runner task ledger unavailable") return False @app.post("/v1/tasks/execute") def execute_task(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return jsonify({"error": "invalid task request"}), 400 node = payload.get("node") parameters = payload.get("parameters", {}) if not isinstance(node, dict) or not isinstance(parameters, dict): return jsonify({"error": "invalid task request"}), 400 replay_only = False try: claims = verifier.verify(payload.get("task_token"), node=node) except TaskTokenExpired: try: claims = verifier.verify_for_replay( payload.get("task_token"), node=node, ) except TaskTokenInvalid: return jsonify({"error": "task token is invalid"}), 401 replay_only = True except TaskTokenInvalid: return jsonify({"error": "task token is invalid"}), 401 binding = { "task_uid": claims.task_uid, "dataflow_uid": claims.dataflow_uid, "deployment_id": claims.deployment_id, "environment": claims.environment, "workflow_version": claims.workflow_version, "correlation_id": claims.correlation_id, "node_id": claims.node_id, "node_type": claims.node_type, "data_source_uid": node.get("data_source_uid"), "idempotency_key": (node.get("idempotency") or {}).get("key"), } if replay_only: claimed = False else: try: claimed = ledger.claim( claims.jti, binding, expires_at=claims.expires_at, ) except Exception: app.logger.error("runner task ledger unavailable") return jsonify({"error": "task ledger is unavailable"}), 503 if not claimed: try: existing = ledger.get(claims.jti) except Exception: return jsonify({"error": "task ledger is unavailable"}), 503 if ( existing is None or any( existing.binding.get(key) != value for key, value in binding.items() ) ): status = 401 if replay_only else 409 return jsonify({"error": "task token already consumed"}), status if existing.status == "running": recovered = None if node.get("type") in {"rule.apply", "quality.check"}: try: recovered = registry.replay_task( node, correlation_id=claims.correlation_id, deployment_id=claims.deployment_id, task_jti=claims.jti, ) except NodeExecutionError: return jsonify( {"error": "task ledger is unavailable"} ), 503 recovered_state = ( recovered.get("state") if isinstance(recovered, dict) else None ) if isinstance(recovered, dict): expected_fields = { "running": {"state"}, "missing": {"state"}, "terminal": { "state", "status", "commit_outcome", "result", "result_digest", "evidence_digest", }, }.get(recovered_state) if ( expected_fields is None or set(recovered) != expected_fields ): return jsonify( {"error": "task ledger is unavailable"} ), 503 recovered_result = ( recovered.get("result") if recovered_state == "terminal" and recovered.get("status") == "success" else None ) if ( recovered_state == "terminal" and recovered.get("status") != "success" ): if not finish_safely( claims.jti, status="unknown", commit_outcome=str( recovered.get("commit_outcome") or "unknown" ), safe_detail=( "task execution evidence is terminal" ), ): return jsonify( {"error": "task ledger is unavailable"} ), 503 return jsonify( { "error": ( "task execution outcome is unknown" ) } ), 409 if recovered_state == "terminal": result_digest = str( recovered.get("result_digest") or "" ) evidence_digest = str( recovered.get("evidence_digest") or "" ) if ( not isinstance(recovered_result, dict) or recovered.get("commit_outcome") != recovered_result.get("commit_outcome") or re.fullmatch( r"[0-9a-f]{64}", result_digest, ) is None or re.fullmatch( r"[0-9a-f]{64}", evidence_digest, ) is None ): return jsonify( {"error": "task ledger is unavailable"} ), 503 encoded_result = json.dumps( recovered_result, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") if ( hashlib.sha256(encoded_result).hexdigest() != result_digest ): return jsonify( {"error": "task ledger is unavailable"} ), 503 replay_body = { "task_uid": claims.task_uid, "correlation_id": claims.correlation_id, "result": recovered_result, } output_artifact = recovered_result.get( "output_artifact" ) if isinstance(output_artifact, str): replay_body["output_artifact"] = output_artifact if not finish_safely( claims.jti, status="success", commit_outcome=recovered_result.get( "commit_outcome", "not_applicable", ), safe_detail="task response recovered", replay_http_status=200, replay_body=replay_body, ): return jsonify( {"error": "task ledger is unavailable"} ), 503 response = jsonify(replay_body) response.headers["X-Idempotent-Replay"] = "true" return response if recovered_state in {None, "missing"}: try: existing = ledger.reconcile_running( claims.jti, binding, ) except Exception: return jsonify( {"error": "task ledger is unavailable"} ), 503 if ( existing is not None and existing.status == "unknown" ): return jsonify( { "error": ( "task execution outcome is unknown" ) } ), 409 elif recovered_state != "running": return jsonify( {"error": "task ledger is unavailable"} ), 503 if replay_only: return jsonify( {"error": "expired task is not replayable"} ), 401 response = jsonify( { "task_uid": claims.task_uid, "correlation_id": claims.correlation_id, "status": "running", } ) response.headers["Retry-After"] = "2" return response, 202 if ( existing.replay_body is None or existing.replay_http_status is None or existing.replay_digest is None ): return jsonify({"error": "task token already consumed"}), 409 encoded = json.dumps( existing.replay_body, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") if ( hashlib.sha256(encoded).hexdigest() != existing.replay_digest ): return jsonify({"error": "task ledger is unavailable"}), 503 response = jsonify(existing.replay_body) response.headers["X-Idempotent-Replay"] = "true" return response, existing.replay_http_status try: result = registry.execute( node, parameters, write_authorized=claims.write_authorized, correlation_id=claims.correlation_id, dataflow_uid=claims.dataflow_uid, deployment_id=claims.deployment_id, environment=claims.environment, workflow_version=claims.workflow_version, node_id=claims.node_id, task_jti=claims.jti, ) except NodeExecutionError as exc: recorded = finish_safely( claims.jti, status=( "unknown" if exc.commit_outcome == "unknown" else "failed" ), commit_outcome=exc.commit_outcome, safe_detail=str(exc), ) if not recorded: return jsonify({"error": "task ledger is unavailable"}), 503 return jsonify({"error": str(exc)}), 400 except Exception: finish_safely( claims.jti, status="failed", safe_detail="task execution failed", ) app.logger.exception("runner task failed") return jsonify({"error": "task execution failed"}), 500 commit_outcome = ( result.get("commit_outcome", "not_applicable") if isinstance(result, dict) else "not_applicable" ) response = { "task_uid": claims.task_uid, "correlation_id": claims.correlation_id, "result": result, } if ( isinstance(result, dict) and isinstance(result.get("output_artifact"), str) ): response["output_artifact"] = result["output_artifact"] replay_body = ( response if node.get("type") in {"rule.apply", "quality.check"} else None ) if not finish_safely( claims.jti, status="success", commit_outcome=commit_outcome, safe_detail="task completed", replay_http_status=(200 if replay_body is not None else None), replay_body=replay_body, ): return jsonify({"error": "task ledger is unavailable"}), 503 return jsonify(response) return app