"""Minimal HTTP surface for the independent DataOps Runner.""" import hashlib import json 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 ) recovered_result = ( recovered.get("result") if recovered_state == "terminal" and recovered.get("status") == "success" else recovered ) 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 ( isinstance(recovered_result, dict) and recovered_state != "missing" ): 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 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