sokuto_server/handlers/
dag.rs

1//! # DAG 実行エンドポイントモジュール
2//!
3//! `POST /v1/systemone/dag` に対する宣言的 DAG リクエスト検証、
4//! インプロセスオーケストレーション実行、および実行経路・決定結果の返却を提供する。
5
6use std::sync::Arc;
7use std::time::Instant;
8
9use axum::extract::State;
10use axum::http::StatusCode;
11use axum::response::IntoResponse;
12use sokuto_runtime::dag::DagError;
13use sokuto_runtime::dag::schema::{DagDefinition, DagNodeType};
14
15use crate::error::{ErrorResponse, ServerError};
16use crate::metrics::{
17    record_dag_execution, record_dag_node_execution, record_http_duration, record_http_request,
18};
19use crate::schema::dag::{DagRequest, DagResponse};
20use crate::state::AppState;
21
22/// DAG 実行タイムアウトのサーバー側ハード上限 (ミリ秒)。
23///
24/// クライアントからの過大なタイムアウト指定によるセッションプールの長期占有や
25/// リソース枯渇 (DoS 脆弱性) を防止する。既定値は 5,000ms (5秒)。
26pub const MAX_DAG_TIMEOUT_MS: u64 = 5000;
27
28/// 宣言的マイクロ決定グラフ (DAG) 推論実行エンドポイント。
29///
30/// 反事実ルールや多段階の条件付き決定グラフを非自己回帰型モデルで順次実行し、
31/// 辿った実行経路、各ステップの所要時間、および最終決定結果を一括返却する。
32/// エスカレーションノードに到達した場合は `status: "escalated"` として正常終了 (200 OK) する。
33#[utoipa::path(
34    post,
35    path = "/v1/systemone/dag",
36    tag = "Inference",
37    request_body = DagRequest,
38    responses(
39        (status = 200, description = "DAG 実行成功 (完了またはエスカレーション)", body = DagResponse),
40        (status = 400, description = "DAG 定義スキーマまたは整合性検証違反", body = ErrorResponse),
41        (status = 504, description = "DAG 実行タイムアウト", body = ErrorResponse),
42        (status = 503, description = "セッションプール枯渇", body = ErrorResponse),
43        (status = 500, description = "推論エンジンまたはサーバー内部障害", body = ErrorResponse)
44    )
45)]
46pub async fn dag_handler(
47    State(state): State<Arc<AppState>>,
48    axum::Json(req): axum::Json<DagRequest>,
49) -> Result<impl IntoResponse, ServerError> {
50    let start_time = Instant::now();
51
52    // 1. State テキストの正規化
53    let state_text = match &req.state {
54        serde_json::Value::String(s) => s.clone(),
55        other => other.to_string(),
56    };
57
58    // 2. DagDefinition DTO からランタイム型への変換 (ハードタイムアウト上限クランプ適用)
59    let mut dag_def: DagDefinition = req.dag.into();
60    let requested_timeout = req.timeout_ms.unwrap_or(dag_def.timeout_ms);
61    dag_def.timeout_ms = requested_timeout.min(MAX_DAG_TIMEOUT_MS);
62
63    let dag_id = dag_def.dag_id.clone();
64
65    // 3. インプロセス DAG 実行器の呼び出し
66    let result = match state.dag_executor.execute(&state_text, &dag_def).await {
67        Ok(res) => res,
68        Err(err) => {
69            let status_code = match &err {
70                DagError::Validation(_)
71                | DagError::StepLimitExceeded(_)
72                | DagError::NodeNotFound(_) => 400,
73                DagError::Timeout(_) => 504,
74                DagError::Pool(_) => 503,
75                _ => 500,
76            };
77            record_http_request("/v1/systemone/dag", status_code);
78            record_http_duration("/v1/systemone/dag", start_time.elapsed().as_secs_f64());
79            return Err(ServerError::Dag(err));
80        }
81    };
82
83    // 4. メトリクス記録
84    let status_str = if result.escalated {
85        "escalated"
86    } else {
87        "completed"
88    };
89    record_dag_execution(&dag_id, status_str, result.total_duration_ms / 1000.0);
90
91    for (node_id, step) in &result.steps {
92        let node_type_str = match step.node_type {
93            DagNodeType::Inference => "inference",
94            DagNodeType::Branch => "branch",
95            DagNodeType::Parallel => "parallel",
96            DagNodeType::Escalate => "escalate",
97        };
98        record_dag_node_execution(&dag_id, node_id, node_type_str, step.duration_ms / 1000.0);
99    }
100
101    record_http_request("/v1/systemone/dag", 200);
102    record_http_duration("/v1/systemone/dag", start_time.elapsed().as_secs_f64());
103
104    // 5. レスポンス構築
105    let response: DagResponse = result.into();
106    Ok((StatusCode::OK, axum::Json(response)))
107}
sokuto_server/handlers/dag.rs

sokuto_server/handlers/
dag.rs
