package com.ai.trackwalker.demo;

import com.ai.trackwalker.api.dto.Requests;
import com.ai.trackwalker.common.Ids;
import com.ai.trackwalker.common.Jsonb;
import com.ai.trackwalker.service.PlatformService;
import jakarta.annotation.PreDestroy;
import org.springframework.http.HttpStatus;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Service
public class DemoRunService {
    private final PlatformService platformService;
    private final JdbcTemplate jdbc;
    private final DemoRunEventPublisher eventPublisher;
    private final ExecutorService executor = Executors.newCachedThreadPool();

    public DemoRunService(PlatformService platformService, JdbcTemplate jdbc, DemoRunEventPublisher eventPublisher) {
        this.platformService = platformService;
        this.jdbc = jdbc;
        this.eventPublisher = eventPublisher;
    }

    public Map<String, Object> startRun() {
        String runId = Ids.next("run");
        Instant now = Instant.now();
        jdbc.update("insert into demo_runs(id, status, current_step, progress, summary, step_snapshots, started_at) values (?,?,?,?,?::jsonb,?::jsonb,?)",
                runId, "running", DemoStep.INITIALIZING.key(), 0, "{}", "{}", Timestamp.from(now));
        eventPublisher.emit(runId, DemoStep.INITIALIZING, "running", 1, "运行实例已创建",
                "正在锁定线路、场景、模型与规则配置", "demo_run", runId,
                Map.of("stage_count", DemoStep.timeline().size(), "dependency_ready", 0),
                initializationPayload(runId, "checking"));
        executor.submit(() -> executeRun(runId, DemoStep.INITIALIZING));
        return Map.of(
                "run_id", runId,
                "status", "running",
                "event_stream_url", "/api/v1/demo/runs/" + runId + "/events",
                "snapshot_url", "/api/v1/demo/runs/" + runId
        );
    }

    public List<Map<String, Object>> listRuns() {
        return jdbc.queryForList("""
                select id as run_id, status, current_step, progress, task_id, summary::text as summary,
                       started_at, completed_at, error_message, next_sequence as event_count
                from demo_runs
                order by started_at desc
                limit 200
                """);
    }

    public Map<String, Object> snapshot(String runId) {
        Map<String, Object> run = jdbc.queryForMap("select *, summary::text as summary_text, step_snapshots::text as step_snapshots_text from demo_runs where id=?", runId);
        List<Map<String, Object>> events = eventPublisher.history(runId);
        Map<String, Object> stepSnapshots = Jsonb.map(String.valueOf(run.get("step_snapshots_text")));
        List<Map<String, Object>> steps = DemoStep.timeline().stream()
                .map(step -> stepStatus(step, stepSnapshots, events))
                .toList();

        Map<String, Object> result = new LinkedHashMap<>();
        result.put("run_id", run.get("id"));
        result.put("status", run.get("status"));
        result.put("current_step", valueOrEmpty(run.get("current_step")));
        result.put("progress", run.get("progress"));
        result.put("task_id", valueOrEmpty(run.get("task_id")));
        result.put("summary", Jsonb.map(String.valueOf(run.get("summary_text"))));
        result.put("started_at", timestampValue(run.get("started_at")));
        result.put("completed_at", timestampValue(run.get("completed_at")));
        result.put("error_message", valueOrEmpty(run.get("error_message")));
        result.put("last_sequence", run.get("next_sequence"));
        result.put("steps", steps);
        result.put("step_snapshots", stepSnapshots);
        result.put("entity_index", entities(runId));
        result.put("events", events);
        return result;
    }

    public Map<String, Object> stepSnapshot(String runId, String stepKey) {
        DemoStep.fromKey(stepKey);
        Map<String, Object> run = jdbc.queryForMap("select step_snapshots::text as snapshots from demo_runs where id=?", runId);
        Map<String, Object> snapshots = Jsonb.map(String.valueOf(run.get("snapshots")));
        Object snapshot = snapshots.getOrDefault(stepKey, Map.of(
                "step_key", stepKey,
                "status", "pending",
                "step_progress", 0,
                "visual_payload", Map.of()
        ));
        List<Map<String, Object>> events = history(runId).stream()
                .filter(event -> stepKey.equals(String.valueOf(event.get("step_key"))))
                .toList();
        return Map.of("run_id", runId, "snapshot", snapshot, "events", events);
    }

    public Map<String, Object> entities(String runId) {
        Map<String, Object> run = jdbc.queryForMap("select task_id from demo_runs where id=?", runId);
        String taskId = valueOrEmpty(run.get("task_id"));
        if (taskId.isBlank()) {
            return Map.of(
                    "task_ids", List.of(), "resource_ids", List.of(), "analysis_job_ids", List.of(),
                    "ai_result_ids", List.of(), "alarm_ids", List.of(), "workorder_ids", List.of(), "sample_ids", List.of()
            );
        }
        return map(
                "task_ids", List.of(taskId),
                "resource_ids", ids("select id from inspection_resources where task_id=? order by created_at", taskId),
                "analysis_job_ids", ids("select id from analysis_jobs where task_id=? order by created_at", taskId),
                "ai_result_ids", ids("select id from ai_results where task_id=? order by created_at", taskId),
                "alarm_ids", ids("select id from alarms where task_id=? order by created_at", taskId),
                "workorder_ids", ids("select wo.id from work_orders wo join alarms a on a.id=wo.alarm_id where a.task_id=? order by wo.created_at", taskId),
                "sample_ids", ids("select id from sample_feedback_records where run_id=? order by created_at", runId)
        );
    }

    public Map<String, Object> exportRun(String runId) {
        return Map.of(
                "schema_version", "1.0",
                "generated_at", Instant.now().toString(),
                "run", snapshot(runId)
        );
    }

    public Map<String, Object> retryStep(String runId, String stepKey, Requests.RetryStepRequest request) {
        DemoStep step;
        try {
            step = DemoStep.fromKey(stepKey);
        } catch (IllegalArgumentException error) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "未知运行节点");
        }
        if (step == DemoStep.RUN_FAILED || step == DemoStep.RUN_CANCELLED) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "终态事件不能作为重试起点");
        }
        Map<String, Object> run = jdbc.queryForMap("select status, progress, task_id, step_snapshots::text as snapshots from demo_runs where id=?", runId);
        String status = String.valueOf(run.get("status"));
        if ("running".equals(status)) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "当前运行仍在执行中");
        }
        Map<String, Object> snapshots = Jsonb.map(String.valueOf(run.get("snapshots")));
        Map<String, Object> current = objectMap(snapshots.get(stepKey));
        int currentAttempt = number(current.get("attempt"), 1);
        if (request != null && request.expectedAttempt() != null && request.expectedAttempt() != currentAttempt) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "节点重试次数已变化，请刷新后重试");
        }

        jdbc.update("update demo_runs set status='running', current_step=?, cancel_requested=false, completed_at=null, error_message=null where id=?",
                step.key(), runId);
        eventPublisher.emitRetry(runId, step, Math.max(0, step.progress() - 1), "节点重试已受理",
                request == null || request.reason() == null ? "正在重新执行所选节点及后续节点" : request.reason(),
                "demo_run", runId,
                Map.of("previous_attempt", currentAttempt),
                map("type", "retry_status", "schema_version", "1.0", "requested_by",
                        request == null || request.operator() == null ? "巡检平台" : request.operator(),
                        "idempotency_key", request == null ? "" : valueOrEmpty(request.idempotencyKey())));
        executor.submit(() -> executeRun(runId, step));
        return Map.of("run_id", runId, "status", "running", "resumed_from", step.key(), "attempt", currentAttempt + 1);
    }

    public List<Map<String, Object>> history(String runId) {
        return eventPublisher.history(runId);
    }

    public SseEmitter subscribe(String runId) {
        return eventPublisher.subscribe(runId);
    }

    public Map<String, Object> cancel(String runId) {
        int updated = jdbc.update("update demo_runs set cancel_requested=true where id=? and status='running'", runId);
        return Map.of("run_id", runId, "cancel_requested", updated > 0);
    }

    @PreDestroy
    public void shutdown() {
        executor.shutdownNow();
    }

    private void executeRun(String runId, DemoStep startStep) {
        String taskId = existingTaskId(runId);
        try {
            if (shouldRun(startStep, DemoStep.INITIALIZING)) {
                pause();
                ensureActive(runId);
                eventPublisher.emit(runId, DemoStep.INITIALIZING, "completed", DemoStep.INITIALIZING.progress(), "运行初始化完成",
                        "线路、场景、能力和关键依赖检查通过", "demo_run", runId,
                        Map.of("stage_count", DemoStep.timeline().size(), "dependency_ready", 6),
                        initializationPayload(runId, "ready"));
            }

            if (shouldRun(startStep, DemoStep.TASK_CREATED)) {
                eventPublisher.emit(runId, DemoStep.TASK_CREATED, "running", 6, "正在创建巡检任务",
                        "校验线路范围、飞行要求、采集能力和场景配置", "inspection_task", "",
                        Map.of("scene_count", scenes().size()),
                        taskPayload("", "validating"));
                Map<String, Object> task = platformService.createTask(new Requests.CreateTaskRequest(
                        "RUN-" + runId.substring(Math.max(0, runId.length() - 8)),
                        "line-demo", "K123+000", "K130+000", "route-demo", scenes(), "high", Instant.now().plusSeconds(300).toString()
                ));
                taskId = String.valueOf(task.get("task_id"));
                jdbc.update("update demo_runs set task_id=? where id=?", taskId, runId);
                eventPublisher.emit(runId, DemoStep.TASK_CREATED, "completed", DemoStep.TASK_CREATED.progress(), "巡检任务创建完成",
                        "任务参数已入库并发布任务创建事件", "inspection_task", taskId,
                        Map.of("task_count", 1, "scene_count", scenes().size()),
                        taskPayload(taskId, "created"));
            }
            requireTask(taskId, startStep);

            ensureActive(runId);
            if (shouldRun(startStep, DemoStep.UAV_DISPATCHING)) {
                emitUavDispatch(runId, taskId);
            }

            ensureActive(runId);
            List<Requests.ResourceItem> resources = resourceDefinitions();
            List<String> resourceIds = resourceIds(taskId);
            if (shouldRun(startStep, DemoStep.RESOURCE_INGESTING)) {
                eventPublisher.emit(runId, DemoStep.RESOURCE_INGESTING, "running", 22, "正在接入多源巡检数据",
                        "可见光、红外和防洪栅格资源进入接入队列", "inspection_task", taskId,
                        Map.of("expected_resources", resources.size(), "received_resources", resourceIds.size()),
                        resourcePayload(resources, resourceIds, "ingesting"));
                if (resourceIds.isEmpty()) {
                    resourceIds = platformService.ingestResources(new Requests.ResourceCompleteRequest(taskId, resources));
                }
                eventPublisher.emit(runId, DemoStep.RESOURCE_INGESTING, "completed", DemoStep.RESOURCE_INGESTING.progress(), "多源数据接入完成",
                        "资源传输、哈希校验、任务关联和对象存储登记完成", "inspection_task", taskId,
                        Map.of("expected_resources", resources.size(), "received_resources", resourceIds.size(), "failed_resources", 0),
                        resourcePayload(resources, resourceIds, "completed"));
            }
            if (resourceIds.isEmpty()) {
                throw new IllegalStateException("未找到可供后续节点使用的数据资源");
            }

            ensureActive(runId);
            if (shouldRun(startStep, DemoStep.PREPROCESS_RUNNING)) {
                emitPreprocess(runId, taskId, resourceIds);
            }

            ensureActive(runId);
            String jobId = latestJobId(taskId);
            if (shouldRun(startStep, DemoStep.AI_INFERENCING)) {
                cleanupPartialAnalysis(runId, taskId);
                eventPublisher.emit(runId, DemoStep.AI_INFERENCING, "running", 44, "正在创建分析任务并加载模型",
                        "视觉、红外和点云模型开始并行处理标准化资源", "inspection_task", taskId,
                        Map.of("resource_count", resourceIds.size(), "model_count", 3, "processed", 0),
                        inferencePayload(taskId, "running", List.of(), Map.of()));
                Map<String, Object> job = platformService.createAnalysisJob(new Requests.CreateAnalysisJobRequest(taskId, resourceIds, scenes(), "offline", "high"));
                jobId = String.valueOf(job.get("analysis_job_id"));
                Map<String, Object> analysis = platformService.runAnalysis(jobId);
                List<Map<String, Object>> results = aiResultsByTask(taskId);
                eventPublisher.emit(runId, DemoStep.AI_INFERENCING, "completed", DemoStep.AI_INFERENCING.progress(), "AI推理分析完成",
                        "模型结果、空间位置和版本信息已持久化", "analysis_job", jobId,
                        Map.of("ai_results", results.size(), "avg_confidence", avgConfidence(taskId), "model_count", 3),
                        inferencePayload(taskId, "completed", results, objectMap(analysis.get("summary"))));
            }

            ensureActive(runId);
            List<Map<String, Object>> alarms = alarmsByTask(taskId, true);
            if (shouldRun(startStep, DemoStep.RULE_EVALUATING)) {
                eventPublisher.emit(runId, DemoStep.RULE_EVALUATING, "running", 60, "正在执行 GIS 与业务规则",
                        "计算线路缓冲区、重点区域、温度和形变阈值", "inspection_task", taskId,
                        Map.of("candidates", count("ai_results where task_id=?", taskId)),
                        rulePayload(alarms, "running"));
                pause();
                eventPublisher.emit(runId, DemoStep.RULE_EVALUATING, "completed", DemoStep.RULE_EVALUATING.progress(), "GIS规则判定完成",
                        "候选结果已形成命中、抑制或无效结论", "inspection_task", taskId,
                        Map.of("rule_hits", ruleHitCount(alarms), "suppressed", suppressedCount(alarms)),
                        rulePayload(alarms, "completed"));
            }

            if (shouldRun(startStep, DemoStep.ALARM_GENERATING)) {
                eventPublisher.emit(runId, DemoStep.ALARM_GENERATING, "completed", DemoStep.ALARM_GENERATING.progress(), "告警生成完成",
                        "有效告警已分级并建立资源、模型和规则证据链", "inspection_task", taskId,
                        Map.of("alarms", activeAlarmCount(alarms), "severity", severityCounts(taskId), "suppressed", suppressedCount(alarms)),
                        alarmPayload(alarms));
            }

            ensureActive(runId);
            List<Map<String, Object>> workorders = workordersByTask(taskId);
            if (shouldRun(startStep, DemoStep.WORKORDER_DISPATCHING)) {
                eventPublisher.emit(runId, DemoStep.WORKORDER_DISPATCHING, "running", 80, "正在派发处置工单",
                        "正在按线路、专业和告警等级匹配责任人员", "inspection_task", taskId,
                        Map.of("workorders", workorders.size(), "receipts", 0),
                        workorderDispatchPayload(workorders, "dispatching"));
                pause();
                eventPublisher.emit(runId, DemoStep.WORKORDER_DISPATCHING, "completed", DemoStep.WORKORDER_DISPATCHING.progress(), "工单派发完成",
                        "工单已送达专业复核人员并收到派发回执", "inspection_task", taskId,
                        Map.of("workorders", workorders.size(), "pending", pendingCount(workorders), "receipts", workorders.size()),
                        workorderDispatchPayload(workorders, "delivered"));
            }

            ensureActive(runId);
            if (shouldRun(startStep, DemoStep.WORKORDER_CLOSING)) {
                int closed = closeSampleWorkorders(workorders);
                workorders = workordersByTask(taskId);
                eventPublisher.emit(runId, DemoStep.WORKORDER_CLOSING, "completed", DemoStep.WORKORDER_CLOSING.progress(), "工单闭环状态已更新",
                        "处置反馈、复核结论和告警状态已同步", "inspection_task", taskId,
                        Map.of("closed", closedCount(workorders), "pending", pendingCount(workorders), "closed_this_run", closed),
                        workorderClosurePayload(workorders, closed));
            }

            ensureActive(runId);
            if (shouldRun(startStep, DemoStep.SAMPLE_FEEDBACK)) {
                List<Map<String, Object>> samples = createSamples(runId, taskId);
                eventPublisher.emit(runId, DemoStep.SAMPLE_FEEDBACK, "completed", DemoStep.SAMPLE_FEEDBACK.progress(), "样本回流已生成",
                        "低置信度、复核确认和场景稀缺样本已进入标注队列", "inspection_task", taskId,
                        Map.of("candidate_samples", samples.size(), "labeling_queue", queuedSamples(samples), "excluded", excludedSamples(samples)),
                        samplePayload(samples));
            }

            Map<String, Object> summary = summaryForTask(runId, taskId);
            jdbc.update("update demo_runs set summary=?::jsonb where id=?", Jsonb.write(summary), runId);
            eventPublisher.emit(runId, DemoStep.RUN_COMPLETED, "completed", DemoStep.RUN_COMPLETED.progress(), "全链路运行完成",
                    "任务、数据、分析、规则、告警、工单和样本记录已完成关联", "inspection_task", taskId,
                    summary,
                    summaryPayload(runId, taskId, summary));
        } catch (CancelledRunException ignored) {
            // Cancellation events are emitted by ensureActive.
        } catch (Exception error) {
            emitFailure(runId, error);
        }
    }

    private void emitUavDispatch(String runId, String taskId) {
        List<Map<String, Object>> dispatchSteps = missionSteps(2);
        eventPublisher.emit(runId, DemoStep.UAV_DISPATCHING, "running", 12, "航线参数校验完成",
                "航线任务正在提交至无人机平台", "inspection_task", taskId,
                Map.of("route_progress", 0, "dispatch_step", 2),
                missionPayload(taskId, "dispatching", 0, "K123+000", dispatchSteps));
        pause();
        dispatchSteps = missionSteps(5);
        eventPublisher.emit(runId, DemoStep.UAV_DISPATCHING, "running", 15, "航线任务已受理并开始飞行",
                "无人机在线，RTK 固定解，正在沿巡检区段采集", "inspection_task", taskId,
                Map.of("route_progress", 68, "battery_percent", 76, "dispatch_step", 5),
                missionPayload(taskId, "flying", 68, "K126+800", dispatchSteps));
        pause();
        eventPublisher.emit(runId, DemoStep.UAV_DISPATCHING, "completed", DemoStep.UAV_DISPATCHING.progress(), "航线任务完成",
                "无人机已返航，任务完成回执和采集点状态已确认", "inspection_task", taskId,
                Map.of("route_progress", 100, "battery_percent", 58, "dispatch_step", 8),
                missionPayload(taskId, "completed", 100, "K130+000", missionSteps(8)));
    }

    private void emitPreprocess(String runId, String taskId, List<String> resourceIds) {
        List<Map<String, Object>> processors = preprocessors(2);
        eventPublisher.emit(runId, DemoStep.PREPROCESS_RUNNING, "running", 31, "格式解析与时间关联完成",
                "图像、红外和栅格资源已完成解码及任务时间关联", "inspection_task", taskId,
                Map.of("finished_processors", 2, "processor_count", 6, "resource_count", resourceIds.size()),
                preprocessPayload(processors, resourceIds, "running"));
        pause();
        processors = preprocessors(4);
        eventPublisher.emit(runId, DemoStep.PREPROCESS_RUNNING, "running", 35, "坐标归一与专项处理完成",
                "WGS84 坐标、线路里程和不同类型处理分支已完成", "inspection_task", taskId,
                Map.of("finished_processors", 4, "processor_count", 6, "resource_count", resourceIds.size()),
                preprocessPayload(processors, resourceIds, "running"));
        pause();
        eventPublisher.emit(runId, DemoStep.PREPROCESS_RUNNING, "completed", DemoStep.PREPROCESS_RUNNING.progress(), "数据预处理完成",
                "质量检查通过，标准化资源已进入分析队列", "inspection_task", taskId,
                Map.of("finished_processors", 6, "processor_count", 6, "quality_pass_rate", 100),
                preprocessPayload(preprocessors(6), resourceIds, "completed"));
    }

    private Map<String, Object> initializationPayload(String runId, String status) {
        List<Map<String, Object>> dependencies = List.of(
                dependency("PostgreSQL/PostGIS", status), dependency("MinIO 对象存储", status),
                dependency("Kafka 事件总线", status), dependency("视觉推理服务", status),
                dependency("点云分析服务", status), dependency("GIS 规则服务", status)
        );
        return map(
                "type", "run_initialization", "schema_version", "1.0",
                "run", map("run_id", runId, "triggered_by", "巡检调度中心", "trigger_mode", "manual", "stage_count", 12),
                "line", map("line_id", "line-demo", "line_name", "铁路重点巡检区段", "mileage_start", "K123+000", "mileage_end", "K130+000", "coordinate_system", "WGS84"),
                "scenes", scenes(), "model_groups", List.of("vision-detector", "thermal-analyzer", "pointcloud-analyzer"),
                "rule_set_version", "rail-rules-2026.07", "dependencies", dependencies,
                "gis_layers", map("railway", true, "mileage_marks", true, "inspection_range", true)
        );
    }

    private Map<String, Object> taskPayload(String taskId, String status) {
        return map(
                "type", "task_detail", "schema_version", "1.0",
                "task", map("task_id", taskId, "task_name", "铁路重点区段无人机综合巡检", "task_type", "comprehensive", "source", "巡检调度中心", "priority", "high", "status", status,
                        "line_id", "line-demo", "direction", "上行", "mileage_start", "K123+000", "mileage_end", "K130+000", "distance_km", 7.0,
                        "planned_start_time", Instant.now().plusSeconds(300).toString(), "planned_end_time", Instant.now().plusSeconds(2100).toString()),
                "flight_requirements", map("flight_height_m", 80, "flight_speed_mps", 8, "battery_return_threshold", 25, "image_overlap", 75, "sampling_interval_s", 2),
                "capture_requirements", List.of(
                        map("type", "image", "required", true, "quality", "4K/清晰度达标"),
                        map("type", "thermal", "required", true, "quality", "温度矩阵完整"),
                        map("type", "tif", "required", true, "quality", "包含地理参考")
                ),
                "scene_set", scenes(), "persistence_status", status
        );
    }

    private Map<String, Object> missionPayload(String taskId, String status, int progress, String mileage, List<Map<String, Object>> steps) {
        return map(
                "type", "uav_dispatch", "schema_version", "1.0", "route_progress", progress, "uav_status", status,
                "mission", map("mission_id", "mission-" + taskId, "external_mission_id", "DJI-" + taskId, "route_id", "route-demo", "vendor", "DJI Cloud API", "status", status,
                        "route_length_km", 8.6, "estimated_minutes", 24, "current_mileage", mileage, "last_callback_at", Instant.now().toString()),
                "uav", map("uav_id", "UAV-M350-01", "model", "Matrice 350 RTK", "payload", "可见光/红外/测绘", "online", true, "battery_percent", progress == 100 ? 58 : 76, "rtk_status", "fixed", "link_quality", 96),
                "position", map("mileage", mileage, "longitude", 116.1 + progress * 0.00002, "latitude", 39.1 + progress * 0.00001),
                "waypoints", List.of(
                        map("index", 1, "mileage", "K123+000", "status", "completed"),
                        map("index", 2, "mileage", "K126+800", "status", progress >= 68 ? "completed" : "pending"),
                        map("index", 3, "mileage", "K130+000", "status", progress == 100 ? "completed" : "pending")
                ),
                "dispatch_steps", steps,
                "latest_receipt", map("code", "0", "message", progress == 100 ? "任务完成" : "任务已受理", "received_at", Instant.now().toString())
        );
    }

    private Map<String, Object> resourcePayload(List<Requests.ResourceItem> definitions, List<String> ids, String status) {
        List<Map<String, Object>> resources = new ArrayList<>();
        for (int index = 0; index < definitions.size(); index++) {
            Requests.ResourceItem definition = definitions.get(index);
            Map<String, Object> metadata = definition.metadata() == null ? Map.of() : definition.metadata();
            boolean complete = ids.size() > index;
            resources.add(map(
                    "resource_id", complete ? ids.get(index) : "pending-" + (index + 1),
                    "resource_name", definition.storageUrl().substring(definition.storageUrl().lastIndexOf('/') + 1),
                    "resource_type", definition.resourceType(), "storage_url", definition.storageUrl(),
                    "capture_time", Instant.now().minusSeconds(20 - index * 5).toString(),
                    "size_bytes", metadata.getOrDefault("size_bytes", 18_000_000L),
                    "upload_progress", complete ? 100 : 35 + index * 20,
                    "ingest_status", complete ? "completed" : status,
                    "checksum_status", complete ? "passed" : "checking", "quality_status", complete ? "qualified" : "pending",
                    "mileage", metadata.getOrDefault("mileage", "K12" + (3 + index) + "+" + (240 + index * 180)),
                    "x", 22 + index * 24, "y", 62 - index * 12
            ));
        }
        Map<String, Long> counts = resources.stream().collect(Collectors.groupingBy(row -> String.valueOf(row.get("resource_type")), LinkedHashMap::new, Collectors.counting()));
        return map(
                "type", "resource_ingestion", "schema_version", "1.0", "resources", resources,
                "resource_stats", counts, "expected_bytes", 243_000_000L,
                "received_bytes", "completed".equals(status) ? 243_000_000L : 96_000_000L,
                "ingest_rate_mbps", "completed".equals(status) ? 0 : 38.6,
                "markers", resourceMarkers(resources)
        );
    }

    private Map<String, Object> preprocessPayload(List<Map<String, Object>> processors, List<String> resourceIds, String status) {
        int completed = (int) processors.stream().filter(item -> "completed".equals(item.get("status"))).count();
        return map(
                "type", "preprocess_pipeline", "schema_version", "1.0", "processors", processors,
                "pipelines", List.of(
                        map("name", "图像/视频", "steps", List.of("解码", "畸变校正", "去噪", "尺寸归一", "质量检查")),
                        map("name", "红外", "steps", List.of("温度矩阵解析", "坏点修复", "温标校正", "质量检查")),
                        map("name", "TIF/点云", "steps", List.of("投影解析", "坐标转换", "栅格裁剪", "形变指标", "质量检查"))
                ),
                "queue_metrics", map("total", 6, "completed", completed, "running", "completed".equals(status) ? 0 : 1, "failed", 0),
                "quality_summary", map("clarity", 0.94, "coordinate_validity", 1.0, "time_sync", "passed", "readability", "passed", "pass_rate", completed == 6 ? 100 : 67),
                "outputs", resourceIds.stream().map(id -> map("resource_id", id, "status", status, "coordinate_system", "WGS84", "quality_report", "passed")).toList()
        );
    }

    private Map<String, Object> inferencePayload(String taskId, String status, List<Map<String, Object>> results, Map<String, Object> summary) {
        int total = results.size();
        return map(
                "type", "inference_workspace", "schema_version", "1.0", "analysis_job", map("task_id", taskId, "status", status, "analysis_mode", "offline", "priority", "high"),
                "models", List.of(
                        model("vision-detector", "rail-vision-yolo11-seg", "v2.4.0", status, total == 0 ? 0 : 12, 0.90, "TensorRT", "GPU-0", 42),
                        model("thermal-analyzer", "rail-thermal-anomaly", "v1.8.2", status, total == 0 ? 0 : 3, 0.90, "ONNX Runtime", "GPU-0", 38),
                        model("pointcloud-analyzer", "rail-flood-change", "v1.5.1", status, total == 0 ? 0 : 3, 0.93, "Open3D", "CPU", 36)
                ),
                "results", results.stream().limit(12).toList(),
                "confidence_buckets", map("0.50-0.69", 2, "0.70-0.84", 5, "0.85-1.00", Math.max(total - 7, 0)),
                "performance", map("elapsed_ms", "completed".equals(status) ? 2860 : 820, "throughput_fps", 18.6, "gpu_utilization", 72, "summary", summary),
                "markers", resultMarkers(results)
        );
    }

    private Map<String, Object> rulePayload(List<Map<String, Object>> alarms, String status) {
        List<Map<String, Object>> rules = List.of(
                rule("R-GIS-001", "线路100米范围", "distance_to_track_m", "<=", 100, "spatial"),
                rule("R-FENCE-002", "防护网人员入侵", "inside_protected_area", "=", true, "spatial"),
                rule("R-WIRE-003", "接触网ROI命中", "inside_wire_roi", "=", true, "spatial"),
                rule("R-FLOOD-004", "防洪体积变化", "volume_change_m3", ">=", 1.2, "threshold"),
                rule("R-THERMAL-005", "设备温度异常", "temperature_c", ">=", 80, "threshold")
        );
        return map(
                "type", "rule_evaluation", "schema_version", "1.0", "status", status, "rules", rules,
                "rule_hits", ruleVisuals(alarms),
                "suppressed_items", alarms.stream().filter(row -> Boolean.TRUE.equals(row.get("suppressed"))).limit(8).toList(),
                "spatial_layers", List.of("铁路中心线", "100米缓冲区", "防护区", "接触网ROI", "重点防洪点", "豁免区"),
                "statistics", map("candidates", alarms.size(), "hits", ruleHitCount(alarms), "suppressed", suppressedCount(alarms)),
                "markers", alarmMarkers(alarms)
        );
    }

    private Map<String, Object> alarmPayload(List<Map<String, Object>> alarms) {
        int total = alarms.size();
        int active = activeAlarmCount(alarms);
        return map(
                "type", "alarm_generation", "schema_version", "1.0",
                "funnel", map("ai_candidates", total, "rule_hits", ruleHitCount(alarms), "suppressed", suppressedCount(alarms), "effective_alarms", active, "pending_dispatch", active),
                "severity_counts", severityCountsFromRows(alarms),
                "alarms", alarms.stream().filter(row -> !Boolean.TRUE.equals(row.get("suppressed"))).limit(12).toList(),
                "evidence_summary", map("resource_linked", active, "model_linked", active, "rule_linked", active, "spatial_located", active),
                "markers", alarmMarkers(alarms)
        );
    }

    private Map<String, Object> workorderDispatchPayload(List<Map<String, Object>> workorders, String status) {
        return map(
                "type", "workorder_dispatch", "schema_version", "1.0", "dispatch_status", status,
                "dispatch_flow", map("alarm_generated", workorders.size(), "workorder_created", workorders.size(), "assignment_matched", workorders.size(), "message_sent", "delivered".equals(status) ? workorders.size() : 0, "accepted", 0),
                "workorders", workorders.stream().limit(12).map(this::decorateWorkorder).toList(),
                "assignment_summary", map("专业复核人员", workorders.size(), "匹配依据", "线路+场景专业+值班计划"),
                "receipts", workorders.stream().limit(12).map(row -> map("workorder_id", row.get("workorder_id"), "channel", "巡检工单API", "code", "delivered".equals(status) ? "0" : "PENDING", "message", "delivered".equals(status) ? "已送达" : "发送中", "received_at", Instant.now().toString())).toList(),
                "markers", workorderMarkers(workorders)
        );
    }

    private Map<String, Object> workorderClosurePayload(List<Map<String, Object>> workorders, int closedThisRun) {
        Map<String, Long> statusCounts = workorders.stream().collect(Collectors.groupingBy(row -> String.valueOf(row.get("status")), LinkedHashMap::new, Collectors.counting()));
        return map(
                "type", "workorder_closure", "schema_version", "1.0", "status_counts", statusCounts,
                "swimlanes", map(
                        "pending", workorders.stream().filter(row -> "created".equals(row.get("status"))).limit(8).toList(),
                        "processing", workorders.stream().filter(row -> "processing".equals(row.get("status"))).limit(8).toList(),
                        "reviewing", List.of(),
                        "closed", workorders.stream().filter(row -> "closed".equals(row.get("status"))).limit(8).toList()
                ),
                "workorders", workorders.stream().limit(12).map(this::decorateWorkorder).toList(),
                "operation_records", workOrderOperations(workorders),
                "alarm_sync_results", map("success", closedCount(workorders), "failed", 0),
                "closed_this_run", closedThisRun, "markers", workorderMarkers(workorders)
        );
    }

    private Map<String, Object> samplePayload(List<Map<String, Object>> samples) {
        Map<String, Long> reasons = samples.stream().collect(Collectors.groupingBy(row -> String.valueOf(row.get("feedback_reason")), LinkedHashMap::new, Collectors.counting()));
        return map(
                "type", "sample_feedback", "schema_version", "1.0",
                "funnel", map("ai_results", samples.size(), "candidate_samples", samples.size(), "auto_excluded", excludedSamples(samples), "labeling_queue", queuedSamples(samples), "pooled", 0),
                "samples", samples.stream().limit(16).toList(), "reason_counts", reasons,
                "model_groups", List.of("vision-detector", "thermal-analyzer", "pointcloud-analyzer"),
                "quality_summary", map("evidence_complete", samples.size(), "privacy_passed", samples.size(), "duplicate", 0, "blocked", 0)
        );
    }

    private Map<String, Object> summaryPayload(String runId, String taskId, Map<String, Object> summary) {
        List<Map<String, Object>> alarms = alarmsByTask(taskId, true);
        List<Map<String, Object>> workorders = workordersByTask(taskId);
        return map(
                "type", "run_summary", "schema_version", "1.0", "summary", summary, "business_metrics", summary,
                "quality_metrics", map("resource_completion_rate", 100, "preprocess_pass_rate", 100, "model_success_rate", 100,
                        "alarm_dispatch_rate", workorders.isEmpty() ? 0 : 100, "workorder_closure_rate", percentage(closedCount(workorders), workorders.size())),
                "step_durations", stepDurations(runId), "exceptions", List.of(), "entity_index", entities(runId),
                "markers", alarmMarkers(alarms)
        );
    }

    private List<Requests.ResourceItem> resourceDefinitions() {
        return List.of(
                new Requests.ResourceItem(null, "image", "s3://rail-inspection/runs/visible-k123-240.jpg", null,
                        map("longitude", 116.1000, "latitude", 39.1000, "distance_to_track_m", 72.5, "mileage", "K123+240", "size_bytes", 18_400_000L)),
                new Requests.ResourceItem(null, "thermal", "s3://rail-inspection/runs/thermal-k126-420.tiff", null,
                        map("temperature_c", 86.3, "longitude", 116.1010, "latitude", 39.1010, "mileage", "K126+420", "size_bytes", 31_600_000L)),
                new Requests.ResourceItem(null, "tif", "s3://rail-inspection/runs/flood-k128-860.tif", null,
                        map("volume_change_m3", 1.8, "longitude", 116.1020, "latitude", 39.1020, "mileage", "K128+860", "size_bytes", 193_000_000L))
        );
    }

    private List<Map<String, Object>> preprocessors(int completedCount) {
        List<String> names = List.of("格式识别与解码", "时间戳与任务关联", "坐标转换与里程映射", "图像/红外专项处理", "TIF形变指标解析", "质量检查与标准化输出");
        List<String> qualities = List.of("格式可读", "时间偏差<50ms", "WGS84坐标有效", "清晰度0.94/温标有效", "体积变化指标有效", "质量通过率100%");
        List<Map<String, Object>> result = new ArrayList<>();
        for (int index = 0; index < names.size(); index++) {
            String status = index < completedCount ? "completed" : index == completedCount ? "running" : "pending";
            result.add(map("processor_id", "processor-" + (index + 1), "processor_name", names.get(index), "status", status,
                    "input_count", 3, "output_count", index < completedCount ? 3 : 0, "elapsed_ms", index < completedCount ? 180 + index * 45 : 0,
                    "quality", qualities.get(index), "retryable", true));
        }
        return result;
    }

    private List<Map<String, Object>> missionSteps(int completedCount) {
        List<String> names = List.of("参数校验", "任务提交", "平台受理", "设备确认", "起飞", "巡航采集", "返航", "任务完成");
        List<Map<String, Object>> result = new ArrayList<>();
        for (int index = 0; index < names.size(); index++) {
            result.add(map("index", index + 1, "name", names.get(index), "status", index < completedCount ? "completed" : index == completedCount ? "running" : "pending"));
        }
        return result;
    }

    private List<Map<String, Object>> aiResultsByTask(String taskId) {
        return jdbc.query("""
                        select id as ai_result_id, job_id, resource_id, scene, category, confidence,
                               geometry::text as geometry, measurements::text as measurements,
                               model_name, model_version, created_at
                        from ai_results where task_id=? order by created_at
                        """,
                (rs, index) -> map(
                        "ai_result_id", rs.getString("ai_result_id"), "job_id", rs.getString("job_id"), "resource_id", rs.getString("resource_id"),
                        "scene", rs.getString("scene"), "category", rs.getString("category"), "confidence", rs.getDouble("confidence"),
                        "geometry", Jsonb.map(rs.getString("geometry")), "measurements", Jsonb.map(rs.getString("measurements")),
                        "model_name", rs.getString("model_name"), "model_version", rs.getString("model_version"),
                        "mileage", "K" + (123 + index % 7) + "+" + String.format("%03d", 120 + index * 37), "created_at", rs.getTimestamp("created_at").toInstant().toString()
                ), taskId);
    }

    private List<Map<String, Object>> alarmsByTask(String taskId, boolean includeSuppressed) {
        String predicate = includeSuppressed ? "" : " and suppressed=false";
        return jdbc.query("""
                        select id as alarm_id, task_id, result_id, scene, category, severity, confidence,
                               location::text as location, evidence::text as evidence, rule_hits::text as rule_hits,
                               status, suppressed, suppression_reason, created_at, updated_at
                        from alarms where task_id=?
                        """ + predicate + " order by created_at",
                (rs, index) -> map(
                        "alarm_id", rs.getString("alarm_id"), "task_id", rs.getString("task_id"), "result_id", rs.getString("result_id"),
                        "scene", rs.getString("scene"), "category", rs.getString("category"), "severity", rs.getString("severity"),
                        "confidence", rs.getDouble("confidence"), "location", Jsonb.map(rs.getString("location")), "evidence", Jsonb.map(rs.getString("evidence")),
                        "rule_hits", Jsonb.stringList(rs.getString("rule_hits")), "status", rs.getString("status"), "suppressed", rs.getBoolean("suppressed"),
                        "suppression_reason", valueOrEmpty(rs.getString("suppression_reason")), "created_at", rs.getTimestamp("created_at").toInstant().toString()
                ), taskId);
    }

    private List<Map<String, Object>> workordersByTask(String taskId) {
        return jdbc.queryForList("""
                select wo.id as workorder_id, wo.alarm_id, wo.status, wo.assignee, wo.close_result, wo.comment,
                       wo.created_at, wo.updated_at, a.task_id, a.scene, a.category, a.severity, a.location::text as location
                from work_orders wo join alarms a on a.id=wo.alarm_id
                where a.task_id=? order by wo.updated_at desc
                """, taskId);
    }

    private List<Map<String, Object>> createSamples(String runId, String taskId) {
        List<Map<String, Object>> existing = samplesByRun(runId);
        if (!existing.isEmpty()) {
            return existing;
        }
        List<Map<String, Object>> results = aiResultsByTask(taskId);
        List<Map<String, Object>> alarms = alarmsByTask(taskId, true);
        Map<String, Map<String, Object>> alarmByResult = alarms.stream().collect(Collectors.toMap(row -> String.valueOf(row.get("result_id")), row -> row, (left, right) -> left));
        Instant now = Instant.now();
        for (int index = 0; index < Math.min(results.size(), 16); index++) {
            Map<String, Object> result = results.get(index);
            Map<String, Object> alarm = alarmByResult.get(String.valueOf(result.get("ai_result_id")));
            String sampleId = Ids.next("sample");
            String reason = number(result.get("confidence"), 1.0) < 0.8 ? "低置信度" : index % 3 == 0 ? "复核确认" : "场景稀缺";
            String modelName = String.valueOf(result.get("model_name"));
            String modelGroup = modelName.contains("thermal") ? "thermal-analyzer" : modelName.contains("point") ? "pointcloud-analyzer" : "vision-detector";
            jdbc.update("""
                            insert into sample_feedback_records(id, run_id, task_id, resource_id, ai_result_id, alarm_id, workorder_id,
                                feedback_reason, priority, labeling_status, target_model_group, source_model_version, quality_status, created_at, updated_at)
                            values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                            """,
                    sampleId, runId, taskId, result.get("resource_id"), result.get("ai_result_id"), alarm == null ? null : alarm.get("alarm_id"),
                    alarm == null ? null : workorderIdByAlarm(valueOrEmpty(alarm.get("alarm_id"))), reason, index < 4 ? "high" : "normal", "queued",
                    modelGroup, result.get("model_version"), "passed", Timestamp.from(now), Timestamp.from(now));
        }
        return samplesByRun(runId);
    }

    private List<Map<String, Object>> samplesByRun(String runId) {
        return jdbc.queryForList("""
                select id as sample_id, task_id, resource_id, ai_result_id, alarm_id, workorder_id,
                       feedback_reason, priority, labeling_status, target_model_group, source_model_version,
                       quality_status, created_at, updated_at
                from sample_feedback_records where run_id=? order by created_at
                """, runId);
    }

    private List<Map<String, Object>> ruleVisuals(List<Map<String, Object>> alarms) {
        return alarms.stream().limit(12).map(alarm -> {
            Map<String, Object> location = objectMap(alarm.get("location"));
            List<String> hits = alarm.get("rule_hits") instanceof List<?> list ? list.stream().map(String::valueOf).toList() : List.of();
            return map(
                    "alarm_id", alarm.get("alarm_id"), "ai_result_id", alarm.get("result_id"), "scene", alarm.get("scene"), "severity", alarm.get("severity"),
                    "input_values", location, "threshold", thresholdForScene(String.valueOf(alarm.get("scene"))),
                    "decision", Boolean.TRUE.equals(alarm.get("suppressed")) ? "suppressed" : "hit", "rule_hits", hits,
                    "suppression_reason", alarm.get("suppression_reason"), "elapsed_ms", 14
            );
        }).toList();
    }

    private List<Map<String, Object>> alarmMarkers(List<Map<String, Object>> alarms) {
        List<Map<String, Object>> visible = alarms.stream().filter(row -> !Boolean.TRUE.equals(row.get("suppressed"))).limit(18).toList();
        int total = Math.max(visible.size() - 1, 1);
        List<Map<String, Object>> markers = new ArrayList<>();
        for (int index = 0; index < visible.size(); index++) {
            Map<String, Object> alarm = visible.get(index);
            double railProgress = visible.size() == 1 ? 0.5 : (double) index / total;
            Map<String, Object> location = objectMap(alarm.get("location"));
            markers.add(map(
                    "id", alarm.get("alarm_id"), "alarm_id", alarm.get("alarm_id"), "scene", alarm.get("scene"),
                    "severity", alarm.get("severity"), "status", alarm.get("status"),
                    "mileage", location.getOrDefault("mileage", "K" + (123 + index % 7) + "+456"),
                    "distance", location.getOrDefault("distance_to_track_m", "线路邻近"),
                    "x", Math.max(5, Math.min(94, 8 + railProgress * 84 + (index % 3 - 1) * 2.4)),
                    "y", Math.max(8, Math.min(88, 70 - railProgress * 54 + (index % 2 == 0 ? -9 : 10)))
            ));
        }
        return markers;
    }

    private List<Map<String, Object>> resultMarkers(List<Map<String, Object>> results) {
        int total = Math.max(results.size() - 1, 1);
        List<Map<String, Object>> markers = new ArrayList<>();
        for (int index = 0; index < Math.min(results.size(), 18); index++) {
            Map<String, Object> row = results.get(index);
            double progress = results.size() == 1 ? 0.5 : (double) index / total;
            markers.add(map("id", row.get("ai_result_id"), "scene", row.get("scene"), "severity", "ai", "status", "detected",
                    "mileage", row.get("mileage"), "distance", "AI识别", "x", 9 + progress * 82, "y", 67 - progress * 50 + (index % 2 == 0 ? -7 : 9)));
        }
        return markers;
    }

    private List<Map<String, Object>> resourceMarkers(List<Map<String, Object>> resources) {
        return resources.stream().map(row -> map("id", row.get("resource_id"), "scene", resourceTypeLabel(row.get("resource_type")),
                "severity", "resource", "status", row.get("ingest_status"), "mileage", row.get("mileage"), "distance", "采集资源",
                "x", row.get("x"), "y", row.get("y"))).toList();
    }

    private List<Map<String, Object>> workorderMarkers(List<Map<String, Object>> workorders) {
        int total = Math.max(workorders.size() - 1, 1);
        List<Map<String, Object>> markers = new ArrayList<>();
        for (int index = 0; index < Math.min(workorders.size(), 18); index++) {
            Map<String, Object> row = workorders.get(index);
            double progress = workorders.size() == 1 ? 0.5 : (double) index / total;
            markers.add(map("id", row.get("workorder_id"), "scene", row.get("scene"), "severity", "closed".equals(row.get("status")) ? "low" : row.get("severity"),
                    "status", row.get("status"), "mileage", "K" + (123 + index % 7) + "+" + String.format("%03d", 180 + index * 41), "distance", "工单",
                    "x", 8 + progress * 84, "y", 70 - progress * 54 + (index % 2 == 0 ? -8 : 10)));
        }
        return markers;
    }

    private Map<String, Object> decorateWorkorder(Map<String, Object> row) {
        Map<String, Object> result = new LinkedHashMap<>(row);
        result.put("professional_type", professionalForScene(String.valueOf(row.get("scene"))));
        result.put("team", "铁路巡检专业班组");
        result.put("priority", "critical".equals(row.get("severity")) ? "urgent" : "high");
        result.put("deadline", Instant.now().plusSeconds(7200).toString());
        result.put("dispatch_channel", "巡检工单API");
        result.put("dispatch_status", "delivered");
        result.put("receipt_code", "0");
        return result;
    }

    private List<Map<String, Object>> workOrderOperations(List<Map<String, Object>> workorders) {
        if (workorders.isEmpty()) {
            return List.of();
        }
        List<String> ids = workorders.stream().map(row -> String.valueOf(row.get("workorder_id"))).toList();
        String placeholders = ids.stream().map(ignored -> "?").collect(Collectors.joining(","));
        return jdbc.queryForList("select id, workorder_id, operation_type, operator_name, result, comment, created_at from work_order_operations where workorder_id in (" + placeholders + ") order by created_at desc", ids.toArray());
    }

    private List<Map<String, Object>> stepDurations(String runId) {
        List<Map<String, Object>> events = history(runId);
        List<Map<String, Object>> durations = new ArrayList<>();
        for (DemoStep step : DemoStep.runnableSteps()) {
            List<Map<String, Object>> stepEvents = events.stream().filter(event -> step.key().equals(event.get("step_key"))).toList();
            if (stepEvents.isEmpty()) {
                durations.add(map("step_key", step.key(), "step_name", step.label(), "status", "pending", "elapsed_ms", 0));
                continue;
            }
            Instant start = Instant.parse(String.valueOf(stepEvents.get(0).get("created_at")));
            Instant end = Instant.parse(String.valueOf(stepEvents.get(stepEvents.size() - 1).get("created_at")));
            Map<String, Object> latest = stepEvents.get(stepEvents.size() - 1);
            durations.add(map("step_key", step.key(), "step_name", step.label(), "status", latest.get("status"),
                    "elapsed_ms", Duration.between(start, end).toMillis(), "attempt", latest.get("attempt")));
        }
        return durations;
    }

    private Map<String, Object> stepStatus(Map<String, Object> step, Map<String, Object> snapshots, List<Map<String, Object>> events) {
        String key = String.valueOf(step.get("key"));
        Map<String, Object> snapshot = objectMap(snapshots.get(key));
        List<Map<String, Object>> stepEvents = events.stream().filter(event -> key.equals(event.get("step_key"))).toList();
        long elapsed = 0;
        if (stepEvents.size() > 1) {
            elapsed = Duration.between(
                    Instant.parse(String.valueOf(stepEvents.get(0).get("created_at"))),
                    Instant.parse(String.valueOf(stepEvents.get(stepEvents.size() - 1).get("created_at")))
            ).toMillis();
        }
        return map(
                "key", key, "name", step.get("name"),
                "run_progress", snapshot.getOrDefault("run_progress", 0),
                "step_progress", snapshot.getOrDefault("step_progress", 0),
                "status", snapshot.getOrDefault("status", "pending"),
                "attempt", snapshot.getOrDefault("attempt", 1), "elapsed_ms", elapsed
        );
    }

    private void cleanupPartialAnalysis(String runId, String taskId) {
        jdbc.update("delete from sample_feedback_records where run_id=?", runId);
        jdbc.update("delete from work_order_operations where workorder_id in (select wo.id from work_orders wo join alarms a on a.id=wo.alarm_id where a.task_id=?)", taskId);
        jdbc.update("delete from work_orders where alarm_id in (select id from alarms where task_id=?)", taskId);
        jdbc.update("delete from alarms where task_id=?", taskId);
        jdbc.update("delete from ai_results where task_id=?", taskId);
        jdbc.update("delete from analysis_jobs where task_id=?", taskId);
    }

    private int closeSampleWorkorders(List<Map<String, Object>> workorders) {
        int limit = Math.min(2, (int) workorders.stream().filter(row -> !"closed".equals(row.get("status"))).count());
        int closed = 0;
        for (Map<String, Object> row : workorders) {
            if (closed >= limit) {
                break;
            }
            if ("closed".equals(row.get("status"))) {
                continue;
            }
            platformService.workOrderCallback(new Requests.WorkOrderCallbackRequest(
                    String.valueOf(row.get("alarm_id")), String.valueOf(row.get("workorder_id")), "closed",
                    "现场处置完成", "隐患已清除，复核通过", "复核人员", Instant.now().toString()
            ));
            closed++;
        }
        return closed;
    }

    private void ensureActive(String runId) {
        Map<String, Object> run = jdbc.queryForMap("select cancel_requested, current_step, progress from demo_runs where id=?", runId);
        if (!Boolean.TRUE.equals(run.get("cancel_requested"))) {
            return;
        }
        DemoStep current = safeStep(String.valueOf(run.get("current_step")));
        if (current != DemoStep.RUN_CANCELLED && current != DemoStep.RUN_FAILED && current != DemoStep.RUN_COMPLETED) {
            eventPublisher.emit(runId, current, "cancelled", number(run.get("progress"), current.progress()), "当前节点已取消",
                    "运行取消请求已生效", "demo_run", runId, Map.of(), map("type", "cancelled", "cancelled_step", current.key()));
        }
        eventPublisher.emit(runId, DemoStep.RUN_CANCELLED, "cancelled", 100, "全链路运行已取消",
                "用户取消了当前运行，已完成节点和业务数据予以保留", "demo_run", runId, Map.of(), Map.of("type", "run_cancelled"));
        throw new CancelledRunException();
    }

    private void emitFailure(String runId, Exception error) {
        Map<String, Object> run = jdbc.queryForMap("select current_step, progress from demo_runs where id=?", runId);
        DemoStep current = safeStep(String.valueOf(run.get("current_step")));
        String message = error.getMessage() == null ? error.getClass().getSimpleName() : error.getMessage();
        if (current != DemoStep.RUN_FAILED && current != DemoStep.RUN_CANCELLED && current != DemoStep.RUN_COMPLETED) {
            eventPublisher.emit(runId, current, "failed", number(run.get("progress"), current.progress()), current.label() + "失败",
                    message, "demo_run", runId, Map.of(), map("type", "step_error", "error_code", error.getClass().getSimpleName(), "retryable", true));
        }
        eventPublisher.emit(runId, DemoStep.RUN_FAILED, "failed", 100, "全链路运行失败", message,
                "demo_run", runId, Map.of(), map("type", "run_error", "failed_step", current.key(), "error_code", error.getClass().getSimpleName()));
    }

    private Map<String, Object> summaryForTask(String runId, String taskId) {
        List<Map<String, Object>> workorders = workordersByTask(taskId);
        return map(
                "tasks", 1,
                "resources", count("inspection_resources where task_id=?", taskId),
                "analysis_jobs", count("analysis_jobs where task_id=?", taskId),
                "ai_results", count("ai_results where task_id=?", taskId),
                "alarms", count("alarms where task_id=? and suppressed=false", taskId),
                "suppressed", count("alarms where task_id=? and suppressed=true", taskId),
                "workorders", workorders.size(), "closed_workorders", closedCount(workorders),
                "pending_workorders", pendingCount(workorders), "sample_candidates", samplesByRun(runId).size()
        );
    }

    private List<String> resourceIds(String taskId) {
        return ids("select id from inspection_resources where task_id=? order by created_at", taskId);
    }

    private String latestJobId(String taskId) {
        List<String> values = ids("select id from analysis_jobs where task_id=? order by created_at desc limit 1", taskId);
        return values.isEmpty() ? "" : values.get(0);
    }

    private String existingTaskId(String runId) {
        Object value = jdbc.queryForObject("select task_id from demo_runs where id=?", Object.class, runId);
        return valueOrEmpty(value);
    }

    private void requireTask(String taskId, DemoStep startStep) {
        if (taskId == null || taskId.isBlank()) {
            throw new IllegalStateException("从 " + startStep.label() + " 恢复时缺少巡检任务");
        }
    }

    private boolean shouldRun(DemoStep start, DemoStep current) {
        List<DemoStep> steps = DemoStep.runnableSteps();
        return steps.indexOf(current) >= steps.indexOf(start);
    }

    private List<String> ids(String sql, Object... args) {
        return jdbc.queryForList(sql, String.class, args);
    }

    private String workorderIdByAlarm(String alarmId) {
        if (alarmId.isBlank()) {
            return null;
        }
        List<String> values = jdbc.queryForList("select id from work_orders where alarm_id=? limit 1", String.class, alarmId);
        return values.isEmpty() ? null : values.get(0);
    }

    private int count(String sqlSuffix, String taskId) {
        Integer value = jdbc.queryForObject("select count(*) from " + sqlSuffix, Integer.class, taskId);
        return value == null ? 0 : value;
    }

    private double avgConfidence(String taskId) {
        Number value = jdbc.queryForObject("select coalesce(avg(confidence),0) from ai_results where task_id=?", Number.class, taskId);
        return value == null ? 0 : Math.round(value.doubleValue() * 100.0) / 100.0;
    }

    private int ruleHitCount(List<Map<String, Object>> alarms) {
        return alarms.stream().mapToInt(alarm -> alarm.get("rule_hits") instanceof List<?> list ? list.size() : 0).sum();
    }

    private int suppressedCount(List<Map<String, Object>> alarms) {
        return (int) alarms.stream().filter(row -> Boolean.TRUE.equals(row.get("suppressed"))).count();
    }

    private int activeAlarmCount(List<Map<String, Object>> alarms) {
        return alarms.size() - suppressedCount(alarms);
    }

    private Map<String, Object> severityCounts(String taskId) {
        return jdbc.queryForList("select severity, count(*) as count from alarms where task_id=? and suppressed=false group by severity", taskId)
                .stream().collect(Collectors.toMap(row -> String.valueOf(row.get("severity")), row -> row.get("count"), (left, right) -> left, LinkedHashMap::new));
    }

    private Map<String, Long> severityCountsFromRows(List<Map<String, Object>> alarms) {
        return alarms.stream().filter(row -> !Boolean.TRUE.equals(row.get("suppressed")))
                .collect(Collectors.groupingBy(row -> String.valueOf(row.get("severity")), LinkedHashMap::new, Collectors.counting()));
    }

    private int pendingCount(List<Map<String, Object>> workorders) {
        return (int) workorders.stream().filter(row -> !"closed".equals(String.valueOf(row.get("status")))).count();
    }

    private int closedCount(List<Map<String, Object>> workorders) {
        return (int) workorders.stream().filter(row -> "closed".equals(String.valueOf(row.get("status")))).count();
    }

    private int queuedSamples(List<Map<String, Object>> samples) {
        return (int) samples.stream().filter(row -> "queued".equals(String.valueOf(row.get("labeling_status")))).count();
    }

    private int excludedSamples(List<Map<String, Object>> samples) {
        return (int) samples.stream().filter(row -> "excluded".equals(String.valueOf(row.get("labeling_status")))).count();
    }

    private double percentage(int numerator, int denominator) {
        return denominator == 0 ? 0 : Math.round(numerator * 1000.0 / denominator) / 10.0;
    }

    private Map<String, Object> dependency(String name, String status) {
        return map("name", name, "status", "ready".equals(status) ? "ready" : "checking", "checked_at", Instant.now().toString(), "latency_ms", "ready".equals(status) ? 12 : 0);
    }

    private Map<String, Object> model(String group, String name, String version, String status, int results, double confidence, String runtime, String device, int processed) {
        return map("model_group", group, "model_name", name, "model_version", version, "status", status, "input_count", 3,
                "processed_count", "completed".equals(status) ? 3 : Math.min(2, processed / 20), "result_count", results, "avg_confidence", confidence,
                "runtime", runtime, "device", device, "throughput", processed, "elapsed_ms", "completed".equals(status) ? 960 : 0);
    }

    private Map<String, Object> rule(String id, String name, String field, String operator, Object threshold, String type) {
        return map("rule_id", id, "rule_name", name, "rule_version", "2026.07", "input_field", field, "operator", operator, "threshold", threshold, "rule_type", type, "status", "enabled");
    }

    private Object thresholdForScene(String scene) {
        if (scene.contains("发热")) return "temperature_c >= 80";
        if (scene.contains("防洪")) return "volume_change_m3 >= 1.2";
        if (scene.contains("接触网")) return "inside_wire_roi = true";
        if (scene.contains("人员")) return "inside_protected_area = true";
        return "distance_to_track_m <= 100";
    }

    private String professionalForScene(String scene) {
        if (scene.contains("接触网") || scene.contains("发热")) return "供电";
        if (scene.contains("防洪")) return "工务防洪";
        return "工务安全";
    }

    private String resourceTypeLabel(Object value) {
        return switch (String.valueOf(value)) {
            case "image" -> "可见光资源";
            case "thermal" -> "红外资源";
            case "tif" -> "防洪栅格";
            case "pointcloud" -> "点云资源";
            default -> "数据资源";
        };
    }

    private List<String> scenes() {
        return List.of("违法开挖", "塔吊", "人员入侵", "防洪点精密监测", "接触网异物（高危）", "杆上设备发热");
    }

    private void pause() {
        try {
            Thread.sleep(420);
        } catch (InterruptedException error) {
            Thread.currentThread().interrupt();
        }
    }

    private DemoStep safeStep(String key) {
        try {
            return DemoStep.fromKey(key);
        } catch (IllegalArgumentException ignored) {
            return DemoStep.INITIALIZING;
        }
    }

    private String timestampValue(Object value) {
        return value instanceof Timestamp timestamp ? timestamp.toInstant().toString() : "";
    }

    private String valueOrEmpty(Object value) {
        return value == null ? "" : String.valueOf(value);
    }

    private int number(Object value, int fallback) {
        return value instanceof Number number ? number.intValue() : fallback;
    }

    private double number(Object value, double fallback) {
        return value instanceof Number number ? number.doubleValue() : fallback;
    }

    private Map<String, Object> objectMap(Object value) {
        if (value instanceof Map<?, ?> source) {
            Map<String, Object> result = new LinkedHashMap<>();
            source.forEach((key, item) -> result.put(String.valueOf(key), item));
            return result;
        }
        return new LinkedHashMap<>();
    }

    private static Map<String, Object> map(Object... values) {
        Map<String, Object> result = new LinkedHashMap<>();
        for (int index = 0; index < values.length; index += 2) {
            result.put(String.valueOf(values[index]), values[index + 1]);
        }
        return result;
    }

    private static final class CancelledRunException extends RuntimeException {
    }
}
