AI 학습 데이터와 모델 파일의 전체 흐름 자동화하기

Data Source
     │
     ▼
Data Processing
     │
     ▼
AI Training
     │
     ▼
Model Output
     │
     ▼
Result Storage

데이터 생성
     │
     ▼
데이터 수집
     │
     ▼
전처리 · AI 처리
     │
     ▼
모델 생성
     │
     ├──────────────┐
     ▼              ▼
모델 저장        결과 전달
     │              │
     └──────┬───────┘
            ▼
        다음 작업

Source A ──┐
           │
Source B ──┼──→ Data Collection
           │
Source C ──┘
                  │
                  ▼
             AI Processing

Collected Data
       │
       ▼
Preprocessing
       │
       ▼
Training Server
       │
       ▼
Inference / Analysis

Data Ready
    │
    ▼
Collection Complete
    │
    ▼
Start Processing
    │
    ▼
Processing Complete
    │
    ▼
Start Training

AI Training
     │
     ▼
Model Generated
     │
 ┌───┴───────────┐
 ▼               ▼
Model Storage  Inference Server
                     │
                     ▼
                Result Output
                     │
                     ▼
               Target System

AI Workflow
     │
     ├── Data Collection   ✓
     │
     ├── Processing        ✓
     │
     ├── Model Training    ●
     │
     └── Result Transfer   ○

Workflow Run
      │
      ▼
Stage Status
      │
 ┌────┴─────┐
 ▼          ▼
Completed   Check Required
                │
                ▼
           Run Details
                │
       ┌────────┼────────┐
       ▼        ▼        ▼
     Data     Device    Transfer
                │
                ▼
           Retry Stage
                │
                ▼
          Result Check

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

// InnorixClient.java
public static final String BASE_URL =
        env("INNORIX_BASE_URL", "https://app.innorix.com").replaceAll("/+
quot;, "");
public static final String WORKSPACE_ID = env("INNORIX_WORKSPACE_ID", null);

public static final int STATUS_COMPLETE = 2;
// States the transfer no longer moves out of
public static final Set<Integer> TERMINAL = Set.of(2, 4, 5, 9, 99);
// Terminal states that are not a full success
public static final Set<Integer> NOT_SUCCEEDED = Set.of(4, 5, 9, 99);

private HttpRequest.Builder headers(HttpRequest.Builder builder) {
    builder.header("Content-Type", "application/json")
           .header("Authorization", "Bearer " + session.accessToken());

    // When omitted the account's current workspace is used.
    if (workspaceId != null) builder.header("x-workspace-id", workspaceId);

    return builder;
}

/** Unwraps and returns data from the response. Throws ApiError on failure. */
public Object api(String method, String path, Object body, Map<String, Object> params) {
    Resp response = request(method, path, body, params);
    Object payload = null;

    try {
        payload = Json.parse(response.text());
    } catch (RuntimeException ignored) {
        payload = null;
    }

    if (!response.ok()) {
        Map<String, Object> map = Json.asObj(payload);
        String message = Json.str(map, "message", Json.str(map, "error", "unknown error"));
        throw new ApiError(response.status, message, map);
    }

    return Json.get(payload, "data");
}

/** Use the server flag when present, otherwise fall back to the status code. */
public static boolean isTerminal(Map<String, Object> record) {
    Boolean flag = Json.boolOrNull(record, "isTerminal");
    if (flag != null) return flag;

    Integer status = Json.intOrNull(record, "status");
    return status != null && TERMINAL.contains(status);
}

// innorix-client.js
const BASE_URL = (process.env.INNORIX_BASE_URL
  || "https://app.innorix.com").replace(/\/+$/, "");
const TOKEN = process.env.INNORIX_ACCESS_TOKEN;
const WORKSPACE_ID = process.env.INNORIX_WORKSPACE_ID || null;

export const STATUS_COMPLETE = 2;
export const TERMINAL = new Set([2, 4, 5, 9, 99]);       // complete / error / cancelled / partial / failed
export const NOT_SUCCEEDED = new Set([4, 5, 9, 99]);

export async function api(method, path, body = null, params = null) {
  const url = new URL(BASE_URL + path);

  if (params) {
    for (const [key, value] of Object.entries(params)) {
      if (value === undefined || value === null) continue;
      url.searchParams.set(key, String(value));
    }
  }

  const headers = {
    "Content-Type": "application/json",
    Authorization: `Bearer ${TOKEN}`,
  };

  // When omitted the account's current workspace is used.
  if (WORKSPACE_ID) headers["x-workspace-id"] = WORKSPACE_ID;

  const response = await fetch(url, {
    method,
    headers,
    body: body === null ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(30_000),
  });

  const payload = await response.json().catch(() => ({}));

  if (!response.ok) {
    throw new Error(payload.message || `HTTP ${response.status}`);
  }

  return payload.data;
}

export function isTerminal(detail) {
  return detail.isTerminal !== undefined
    ? detail.isTerminal
    : TERMINAL.has(detail.status);
}

// InnorixClient.cs
public static readonly string BaseUrl =
    Env("INNORIX_BASE_URL", "https://app.innorix.com").TrimEnd('/');
public static readonly string WorkspaceIdFromEnv = Env("INNORIX_WORKSPACE_ID", null);

public const int StatusComplete = 2;
/// <summary>States the transfer no longer moves out of</summary>
public static readonly HashSet<int> Terminal = new HashSet<int> { 2, 4, 5, 9, 99 };
/// <summary>Terminal states that are not a full success</summary>
public static readonly HashSet<int> NotSucceeded = new HashSet<int> { 4, 5, 9, 99 };

// Applied on every request
request.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Session.AccessToken);
// When omitted the account's current workspace is used.
if (WorkspaceId != null) request.Headers.TryAddWithoutValidation("x-workspace-id", WorkspaceId);

public async Task<JsonNode> ApiAsync(string method, string path, JsonNode body = null,
    IDictionary<string, object> parameters = null)
{
    Resp response = await RequestAsync(method, path, body, parameters).ConfigureAwait(false);
    JsonNode payload = null;

    try
    {
        payload = J.Parse(response.Text());
    }
    catch (Exception)
    {
        payload = null;
    }

    if (!response.Ok)
    {
        JsonObject map = J.AsObj(payload);
        string message = J.Str(map, "message", J.Str(map, "error", "unknown error"));
        throw new ApiError(response.Status, message, map);
    }

    return J.Get(payload, "data");
}

/// <summary>Use the server flag when present, otherwise fall back to the status code.</summary>
public static bool IsTerminal(JsonObject record)
{
    bool? flag = J.BoolOrNull(record, "isTerminal");
    if (flag != null) return flag.Value;

    int? status = J.IntOrNull(record, "status");
    return status != null && Terminal.Contains(status.Value);
}

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

public static String encodePath(String deviceId, String rawPath) {
    String normalized = (rawPath == null ? "" : rawPath).replace("\\", "/");
    return deviceId + "_ino_"
            + Base64.getEncoder().encodeToString(normalized.getBytes(StandardCharsets.UTF_8));
}

public static String nowIso() {
    return Instant.now().truncatedTo(ChronoUnit.SECONDS).toString().replace("Z", ".000Z");
}

export function encodePath(deviceId, rawPath) {
  const normalized = String(rawPath ?? "").replace(/\\/g, "/");
  const token = Buffer.from(normalized, "utf8").toString("base64");
  return `${deviceId}_ino_${token}`;
}

export function nowIso() {
  return new Date().toISOString().replace(/\.\d{3}Z$/, ".000Z");
}

public static string EncodePath(string deviceId, string rawPath)
{
    string normalized = (rawPath ?? "").Replace("\\", "/");
    return deviceId + "_ino_" + Convert.ToBase64String(Encoding.UTF8.GetBytes(normalized));
}

public static string NowIso()
{
    return DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss'.000Z'",
        System.Globalization.CultureInfo.InvariantCulture);
}

DATA_ROOT = "/data"
WORK_ROOT = "/work"
ARCHIVE_ROOT = "/archive"


# source dataset on the storage device
def dataset_path(dataset, version):
    return f"{DATA_ROOT}/datasets/{dataset}/{version}"


# path the training job reads on the node
def node_input_path(dataset, version, run_id):
    return f"{WORK_ROOT}/{run_id}/input/{dataset}/{version}"


# path the node writes its results to
def node_output_path(run_id):
    return f"{WORK_ROOT}/{run_id}/output"


# archive path where results are collected
def archive_path(run_id):
    return f"{ARCHIVE_ROOT}/runs/{run_id}"


run_id = f"r-{time.strftime('%Y%m%d-%H%M%S')}"

static final String DATA_ROOT = "/data";
static final String WORK_ROOT = "/work";
static final String ARCHIVE_ROOT = "/archive";

// source dataset on the storage device
String datasetPath(String dataset, String version) {
    return DATA_ROOT + "/datasets/" + dataset + "/" + version;
}

// path the training job reads on the node
String nodeInputPath(String dataset, String version, String runId) {
    return WORK_ROOT + "/" + runId + "/input/" + dataset + "/" + version;
}

// path the node writes its results to
String nodeOutputPath(String runId) {
    return WORK_ROOT + "/" + runId + "/output";
}

// archive path where results are collected
String archivePath(String runId) {
    return ARCHIVE_ROOT + "/runs/" + runId;
}

String runId = "r-" + DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss")
        .withZone(ZoneOffset.UTC).format(Instant.now());

const DATA_ROOT = "/data";
const WORK_ROOT = "/work";
const ARCHIVE_ROOT = "/archive";

// source dataset on the storage device
const datasetPath = (dataset, version) => `${DATA_ROOT}/datasets/${dataset}/${version}`;

// path the training job reads on the node
const nodeInputPath = (dataset, version, runId) =>
  `${WORK_ROOT}/${runId}/input/${dataset}/${version}`;

// path the node writes its results to
const nodeOutputPath = (runId) => `${WORK_ROOT}/${runId}/output`;

// archive path where results are collected
const archivePath = (runId) => `${ARCHIVE_ROOT}/runs/${runId}`;

const runId = `r-${new Date().toISOString().replace(/[-:T]/g, "").slice(0, 15)}`;

const string DataRoot = "/data";
const string WorkRoot = "/work";
const string ArchiveRoot = "/archive";

// source dataset on the storage device
string DatasetPath(string dataset, string version) =>
    
quot;{DataRoot}/datasets/{dataset}/{version}";

// path the training job reads on the node
string NodeInputPath(string dataset, string version, string runId) =>
    
quot;{WorkRoot}/{runId}/input/{dataset}/{version}";

// path the node writes its results to
string NodeOutputPath(string runId) => 
quot;{WorkRoot}/{runId}/output";

// archive path where results are collected
string ArchivePath(string runId) => 
quot;{ArchiveRoot}/runs/{runId}";

string runId = 
quot;r-{DateTime.UtcNow:yyyyMMdd-HHmmss}";

def deploy(storage, nodes, dataset, version, run_id):
    source_path = dataset_path(dataset, version)
    target_path = node_input_path(dataset, version, run_id)

    return {
        node: api("POST", "/api/transfers/manual", {
            "sourceDevice": storage,
            "targetDevice": node,
            "targetPath": target_path,
            "sourcePaths": [source_path],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]
        for node in nodes
    }

Map<String, String> deploy(String storage, List<String> nodes,
                           String dataset, String version, String runId) {
    String sourcePath = datasetPath(dataset, version);
    String targetPath = nodeInputPath(dataset, version, runId);
    Map<String, String> transfers = new LinkedHashMap<>();

    for (String node : nodes) {
        Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
                "sourceDevice", storage,
                "targetDevice", node,
                "targetPath", targetPath,
                "sourcePaths", List.of(sourcePath),
                "sendAllFolder", true,
                "checkIntegrity", true,
                "transferOptions", Json.newObj("target-action", "overwrite")));
        transfers.put(node, Json.str(transfer, "monitorId"));
    }

    return transfers;
}

async function deploy(storage, nodes, dataset, version, runId) {
  const sourcePath = datasetPath(dataset, version);
  const targetPath = nodeInputPath(dataset, version, runId);
  const transfers = {};

  for (const node of nodes) {
    const transfer = await client.api("POST", "/api/transfers/manual", {
      sourceDevice: storage,
      targetDevice: node,
      targetPath,
      sourcePaths: [sourcePath],
      sendAllFolder: true,
      checkIntegrity: true,
      transferOptions: { "target-action": "overwrite" },
    });
    transfers[node] = transfer.monitorId;
  }

  return transfers;
}

async Task<Dictionary<string, string>> DeployAsync(string storage,
    IEnumerable<string> nodes, string dataset, string version, string runId)
{
    string sourcePath = DatasetPath(dataset, version);
    string targetPath = NodeInputPath(dataset, version, runId);
    var transfers = new Dictionary<string, string>();

    foreach (string node in nodes)
    {
        JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
            new JsonObject
            {
                ["sourceDevice"] = storage,
                ["targetDevice"] = node,
                ["targetPath"] = targetPath,
                ["sourcePaths"] = new JsonArray { sourcePath },
                ["sendAllFolder"] = true,
                ["checkIntegrity"] = true,
                ["transferOptions"] = new JsonObject { ["target-action"] = "overwrite" },
            });
        transfers[node] = J.Str(transfer, "monitorId");
    }

    return transfers;
}

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

public Map<String, Object> wait(String monitorId, int timeoutSeconds, int intervalSeconds) {
    long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L;

    while (System.currentTimeMillis() < deadline) {
        Map<String, Object> detail = apiObj("GET", "/api/transfers/" + monitorId);

        if (isTerminal(detail)) return detail;

        sleep(intervalSeconds * 1000L);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public List<Map<String, Object>> failedFiles(String monitorId) {
    List<Map<String, Object>> failed = new ArrayList<>();

    for (Map<String, Object> row : transferFiles(monitorId)) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && NOT_SUCCEEDED.contains(status)) failed.add(row);
    }

    return failed;
}

public int retryFailed(String monitorId) {
    List<Map<String, Object>> rows = failedFiles(monitorId);
    if (rows.isEmpty()) return 0;

    List<Object> filesRetry = new ArrayList<>();

    for (Map<String, Object> row : rows) {
        String path = Json.str(row, "sourceFilePath");
        if (path == null) continue;
        filesRetry.add(Json.newObj("filePath", path, "isDir", Json.bool(row, "isFolder", false)));
    }

    api("POST", "/api/transfers/" + monitorId + "/retry", Json.newObj("filesRetry", filesRetry));

    return rows.size();
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function wait(monitorId, { timeout = 3600, interval = 3 } = {}) {
  const deadline = Date.now() + timeout * 1000;

  while (Date.now() < deadline) {
    const detail = await api("GET", `/api/transfers/${monitorId}`);

    if (isTerminal(detail)) return detail;

    await sleep(interval * 1000);
  }

  throw new Error(`timeout waiting for ${monitorId}`);
}

export async function failedFiles(monitorId) {
  const result = (await api("GET", `/api/transfers/${monitorId}/files`, null, {
    state: "any", size: 500,
  })) || {};

  return (result.children || []).filter((row) => NOT_SUCCEEDED.has(row.status));
}

export async function retryFailed(monitorId) {
  const rows = await failedFiles(monitorId);

  if (rows.length === 0) return 0;

  await api("POST", `/api/transfers/${monitorId}/retry`, {
    filesRetry: rows
      .filter((row) => row.sourceFilePath)
      .map((row) => ({ filePath: row.sourceFilePath, isDir: Boolean(row.isFolder) })),
  });

  return rows.length;
}

public async Task<JsonObject> WaitAsync(string monitorId, int timeoutSeconds = 3600,
    int intervalSeconds = 3)
{
    long deadline = Environment.TickCount64 + timeoutSeconds * 1000L;

    while (Environment.TickCount64 < deadline)
    {
        JsonObject detail = await ApiObjAsync("GET", "/api/transfers/" + monitorId)
            .ConfigureAwait(false);

        if (IsTerminal(detail)) return detail;

        await Task.Delay(intervalSeconds * 1000).ConfigureAwait(false);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public async Task<List<JsonObject>> FailedFilesAsync(string monitorId)
{
    List<JsonObject> rows = await TransferFilesAsync(monitorId).ConfigureAwait(false);

    return rows.Where(row =>
    {
        int? status = J.IntOrNull(row, "status");
        return status != null && NotSucceeded.Contains(status.Value);
    }).ToList();
}

/// <summary>Only callable once the transfer has reached a terminal state.</summary>
public async Task<int> RetryFailedAsync(string monitorId)
{
    List<JsonObject> rows = await FailedFilesAsync(monitorId).ConfigureAwait(false);
    if (rows.Count == 0) return 0;

    var filesRetry = new JsonArray();

    foreach (JsonObject row in rows)
    {
        string path = J.Str(row, "sourceFilePath");
        if (path == null) continue;

        filesRetry.Add(new JsonObject
        {
            ["filePath"] = path,
            ["isDir"] = J.Bool(row, "isFolder", false),
        });
    }

    await ApiAsync("POST", "/api/transfers/" + monitorId + "/retry",
        new JsonObject { ["filesRetry"] = filesRetry }).ConfigureAwait(false);

    return rows.Count;
}

transfers = deploy(STORAGE, NODES, "imagenet", "v3", run_id)
results = {node: wait(mid) for node, mid in transfers.items()}

incomplete = [n for n, d in results.items() if d["status"] != STATUS_COMPLETE]

if incomplete:
    raise RuntimeError(f"nodes not fully delivered: {', '.join(incomplete)}")

Map<String, String> transfers = deploy(STORAGE, NODES, "imagenet", "v3", runId);
Map<String, Map<String, Object>> results = new LinkedHashMap<>();
for (var entry : transfers.entrySet()) {
    results.put(entry.getKey(), client.await(entry.getValue(), 7200, null));
}

List<String> incomplete = new ArrayList<>();
results.forEach((node, detail) -> {
    if (Json.intOr(detail, "status", -1) != InnorixClient.STATUS_COMPLETE) {
        incomplete.add(node);
    }
});

if (!incomplete.isEmpty()) {
    throw new RuntimeException("전달 미완료 노드: " + String.join(", ", incomplete));
}

const transfers = await deploy(STORAGE, NODES, "imagenet", "v3", runId);
const results = {};
for (const [node, mid] of Object.entries(transfers)) {
  results[node] = await wait(mid);
}

const incomplete = Object.entries(results)
  .filter(([, d]) => d.status !== STATUS_COMPLETE)
  .map(([node]) => node);

if (incomplete.length) {
  throw new Error(`전달 미완료 노드: ${incomplete.join(", ")}`);
}

Dictionary<string, string> transfers = await DeployAsync(
    STORAGE, NODES, "imagenet", "v3", runId);
var results = new Dictionary<string, JsonObject>();
foreach (var (node, mid) in transfers)
{
    results[node] = await client.WaitAsync(mid, 7200);
}

var incomplete = results
    .Where(kv => J.IntOrNull(kv.Value, "status") != InnorixClient.StatusComplete)
    .Select(kv => kv.Key).ToList();

if (incomplete.Count > 0)
{
    throw new Exception(
quot;전달 미완료 노드: {string.Join(", ", incomplete)}");
}

monitor_id = api("POST", "/api/transfers/manual", {
    "sourceDevice": STORAGE,
    "targetDevice": node,
    "targetPath": node_input_path("imagenet", "v3", run_id),
    "sourcePaths": [dataset_path("imagenet", "v3")],
    "sendAllFolder": True,
    "incremental": True,
    "transferOptions": {"target-action": "overwrite"},
})["monitorId"]

detail = wait(monitor_id)
print(f"transferred {detail['fileCount']} files, {detail['totalSize']} bytes")

Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
        "sourceDevice", STORAGE,
        "targetDevice", node,
        "targetPath", nodeInputPath("imagenet", "v3", runId),
        "sourcePaths", List.of(datasetPath("imagenet", "v3")),
        "sendAllFolder", true,
        "incremental", true,
        "transferOptions", Json.newObj("target-action", "overwrite")));
String monitorId = Json.str(transfer, "monitorId");

Map<String, Object> detail = client.await(monitorId, 7200, null);
System.out.printf("전송 파일 %s개, %s bytes%n",
        Json.str(detail, "fileCount"), Json.str(detail, "totalSize"));

const transfer = await client.api("POST", "/api/transfers/manual", {
  sourceDevice: STORAGE,
  targetDevice: node,
  targetPath: nodeInputPath("imagenet", "v3", runId),
  sourcePaths: [datasetPath("imagenet", "v3")],
  sendAllFolder: true,
  incremental: true,
  transferOptions: { "target-action": "overwrite" },
});
const monitorId = transfer.monitorId;

const detail = await wait(monitorId);
console.log(`전송 파일 ${detail.fileCount}개, ${detail.totalSize} bytes`);

JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
    new JsonObject
    {
        ["sourceDevice"] = STORAGE,
        ["targetDevice"] = node,
        ["targetPath"] = NodeInputPath("imagenet", "v3", runId),
        ["sourcePaths"] = new JsonArray { DatasetPath("imagenet", "v3") },
        ["sendAllFolder"] = true,
        ["incremental"] = true,
        ["transferOptions"] = new JsonObject { ["target-action"] = "overwrite" },
    });
string monitorId = J.Str(transfer, "monitorId");

JsonObject detail = await client.WaitAsync(monitorId, 7200);
Console.WriteLine(
quot;전송 파일 {J.Str(detail, "fileCount")}개, "
    + 
quot;{J.Str(detail, "totalSize")} bytes");

def verify(monitor_id, timeout=1800, interval=10):
    api("POST", f"/api/transfers/{monitor_id}/verification", {})

    deadline = time.time() + timeout

    while time.time() < deadline:
        result = api("GET", f"/api/transfers/{monitor_id}/verification") or {}

        if result.get("verified"):
            return result

        time.sleep(interval)

    raise TimeoutError(f"verification: {monitor_id}")


result = verify(monitor_id)

if result["sourceFileCount"] != result["targetFileCount"]:
    raise RuntimeError("file counts differ - part of the transfer is missing")

if not result["checksumMatched"]:
    for row in result.get("mismatchedFiles") or []:
        print("mismatch:", row)

Map<String, Object> result = client.verify(monitorId, 1800, 10);

if (!Json.str(result, "sourceFileCount").equals(Json.str(result, "targetFileCount"))) {
    throw new RuntimeException("파일 수가 다릅니다. 전송이 일부 빠졌습니다.");
}

if (!Json.bool(result, "checksumMatched", false)) {
    for (Object row : Json.arrOf(result, "mismatchedFiles")) {
        System.out.println("불일치: " + row);
    }
}

const result = await verify(monitorId);

if (result.sourceFileCount !== result.targetFileCount) {
  throw new Error("파일 수가 다릅니다. 전송이 일부 빠졌습니다.");
}

if (!result.checksumMatched) {
  for (const row of result.mismatchedFiles || []) {
    console.log("불일치:", row);
  }
}

JsonObject result = await client.VerifyAsync(monitorId);

if (J.Str(result, "sourceFileCount") != J.Str(result, "targetFileCount"))
{
    throw new Exception("파일 수가 다릅니다. 전송이 일부 빠졌습니다.");
}

if (!J.Bool(result, "checksumMatched", false))
{
    foreach (JsonNode row in J.ArrOf(result, "mismatchedFiles"))
    {
        Console.WriteLine(
quot;불일치: {row}");
    }
}

def collect(nodes, archive, run_id):
    transfers = {}

    for node in nodes:
        target_path = f"{archive_path(run_id)}/{node}"

        transfers[node] = api("POST", "/api/transfers/manual", {
            "sourceDevice": node,
            "targetDevice": archive,
            "targetPath": target_path,
            "sourcePaths": [node_output_path(run_id)],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]

    return transfers

Map<String, String> collect(List<String> nodes, String archive, String runId) {
    Map<String, String> transfers = new LinkedHashMap<>();

    for (String node : nodes) {
        String targetPath = archivePath(runId) + "/" + node;

        Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
                "sourceDevice", node,
                "targetDevice", archive,
                "targetPath", targetPath,
                "sourcePaths", List.of(nodeOutputPath(runId)),
                "sendAllFolder", true,
                "checkIntegrity", true,
                "transferOptions", Json.newObj("target-action", "overwrite")));
        transfers.put(node, Json.str(transfer, "monitorId"));
    }

    return transfers;
}

async function collect(nodes, archive, runId) {
  const transfers = {};

  for (const node of nodes) {
    const targetPath = `${archivePath(runId)}/${node}`;

    const transfer = await client.api("POST", "/api/transfers/manual", {
      sourceDevice: node,
      targetDevice: archive,
      targetPath,
      sourcePaths: [nodeOutputPath(runId)],
      sendAllFolder: true,
      checkIntegrity: true,
      transferOptions: { "target-action": "overwrite" },
    });
    transfers[node] = transfer.monitorId;
  }

  return transfers;
}

async Task<Dictionary<string, string>> CollectAsync(
    IEnumerable<string> nodes, string archive, string runId)
{
    var transfers = new Dictionary<string, string>();

    foreach (string node in nodes)
    {
        string targetPath = 
quot;{ArchivePath(runId)}/{node}";

        JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
            new JsonObject
            {
                ["sourceDevice"] = node,
                ["targetDevice"] = archive,
                ["targetPath"] = targetPath,
                ["sourcePaths"] = new JsonArray { NodeOutputPath(runId) },
                ["sendAllFolder"] = true,
                ["checkIntegrity"] = true,
                ["transferOptions"] = new JsonObject { ["target-action"] = "overwrite" },
            });
        transfers[node] = J.Str(transfer, "monitorId");
    }

    return transfers;
}

/archive/runs/r-20260901-0200/
    dev-gpu-01/
        model.pt
        metrics.json
    dev-gpu-02/
        model.pt
        metrics.json

import uuid

flow_id = str(uuid.uuid4())


def build_step(name, source, source_path, target, target_path,
               step, trigger_id=None, webhook=None):
    schedule = {
        "type": "none",
        "startDateType": "now",
        "startDate": now_iso(),
        "timezone": "Asia/Seoul",
    }

    if trigger_id:
        schedule["triggerAutomation"] = {"value": trigger_id}

    body = {
        "name": name,
        "flowName": name,
        "flowId": flow_id,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": step,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": step,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": "overwrite",
                    "send-fileoption": {},
                },
            }
        ],
        "schedules": [schedule],
    }

    if webhook:
        body["processors"] = [{
            "category": "run",
            "type": "http",
            "config": {"url": webhook, "method": "POST"},
        }]

    return body


deploy_id = api("POST", "/api/automations", build_step(
    f"deploy imagenet:v3", STORAGE, dataset_path("imagenet", "v3"),
    node, node_input_path("imagenet", "v3", run_id),
    step=1, webhook=TRAIN_HOOK))["automationId"]

collect_id = api("POST", "/api/automations", build_step(
    f"collect {run_id}", node, node_output_path(run_id),
    ARCHIVE, f"{archive_path(run_id)}/{node}",
    step=2, trigger_id=deploy_id))["automationId"]

String flowId = UUID.randomUUID().toString();

Map<String, Object> buildStep(String name, String source, String sourcePath,
                              String target, String targetPath, int step,
                              String triggerId, String webhook) {
    Map<String, Object> schedule = new LinkedHashMap<>(Json.newObj(
            "type", "none", "startDateType", "now",
            "startDate", InnorixClient.nowIso(), "timezone", "Asia/Seoul"));

    if (triggerId != null) {
        schedule.put("triggerAutomation", Json.newObj("value", triggerId));
    }

    Map<String, Object> detail = Json.newObj(
            "senderId", source, "receiverId", target,
            "sourceItem", List.of(Json.newObj(
                    "hash", InnorixClient.encodePath(source, sourcePath),
                    "filePath", sourcePath, "isDir", true)),
            "targetPath", InnorixClient.encodePath(target, targetPath),
            "step", step,
            "transferOptions", Json.newObj(
                    "noSchedule", false, "target-action", "overwrite",
                    "send-fileoption", Json.newObj()));

    Map<String, Object> body = new LinkedHashMap<>(Json.newObj(
            "name", name, "flowName", name, "flowId", flowId,
            "transferType", "normal", "timezone", "Asia/Seoul",
            "step", step, "isUpcoming", false,
            "details", List.of(detail),
            "schedules", List.of(schedule)));

    if (webhook != null) {
        body.put("processors", List.of(Json.newObj(
                "category", "run", "type", "http",
                "config", Json.newObj("url", webhook, "method", "POST"))));
    }

    return body;
}

import { randomUUID } from "node:crypto";

const flowId = randomUUID();

function buildStep(name, source, sourcePath, target, targetPath, {
  step, triggerId = null, webhook = null,
}) {
  const schedule = {
    type: "none", startDateType: "now",
    startDate: nowIso(), timezone: "Asia/Seoul",
  };

  if (triggerId) schedule.triggerAutomation = { value: triggerId };

  const body = {
    name, flowName: name, flowId,
    transferType: "normal", timezone: "Asia/Seoul",
    step, isUpcoming: false,
    details: [{
      senderId: source, receiverId: target,
      sourceItem: [{
        hash: encodePath(source, sourcePath),
        filePath: sourcePath, isDir: true,
      }],
      targetPath: encodePath(target, targetPath),
      step,
      transferOptions: {
        noSchedule: false, "target-action": "overwrite",
        "send-fileoption": {},
      },
    }],
    schedules: [schedule],
  };

  if (webhook) {
    body.processors = [{
      category: "run", type: "http",
      config: { url: webhook, method: "POST" },
    }];
  }

  return body;
}

string flowId = Guid.NewGuid().ToString();

JsonObject BuildStep(string name, string source, string sourcePath,
    string target, string targetPath, int step,
    string triggerId = null, string webhook = null)
{
    var schedule = new JsonObject
    {
        ["type"] = "none", ["startDateType"] = "now",
        ["startDate"] = InnorixClient.NowIso(), ["timezone"] = "Asia/Seoul",
    };

    if (triggerId != null)
        schedule["triggerAutomation"] = new JsonObject { ["value"] = triggerId };

    var detail = new JsonObject
    {
        ["senderId"] = source, ["receiverId"] = target,
        ["sourceItem"] = new JsonArray
        {
            new JsonObject
            {
                ["hash"] = InnorixClient.EncodePath(source, sourcePath),
                ["filePath"] = sourcePath, ["isDir"] = true,
            },
        },
        ["targetPath"] = InnorixClient.EncodePath(target, targetPath),
        ["step"] = step,
        ["transferOptions"] = new JsonObject
        {
            ["noSchedule"] = false, ["target-action"] = "overwrite",
            ["send-fileoption"] = new JsonObject(),
        },
    };

    var body = new JsonObject
    {
        ["name"] = name, ["flowName"] = name, ["flowId"] = flowId,
        ["transferType"] = "normal", ["timezone"] = "Asia/Seoul",
        ["step"] = step, ["isUpcoming"] = false,
        ["details"] = new JsonArray { detail },
        ["schedules"] = new JsonArray { schedule },
    };

    if (webhook != null)
    {
        body["processors"] = new JsonArray
        {
            new JsonObject
            {
                ["category"] = "run", ["type"] = "http",
                ["config"] = new JsonObject { ["url"] = webhook, ["method"] = "POST" },
            },
        };
    }

    return body;
}

def on_train_hook(payload):
    monitor_id = payload.get("monitorId")

    if monitor_id:
        detail = wait(monitor_id)

        if detail["status"] != STATUS_COMPLETE:
            return abort_run(payload)

    start_training(payload)

void onTrainHook(Map<String, Object> payload) throws Exception {
    String monitorId = Json.str(payload, "monitorId");

    if (monitorId != null) {
        Map<String, Object> detail = client.await(monitorId, 7200, null);
        if (Json.intOr(detail, "status", -1) != InnorixClient.STATUS_COMPLETE) {
            abortRun(payload);
            return;
        }
    }

    startTraining(payload);
}

async function onTrainHook(payload) {
  const monitorId = payload.monitorId;

  if (monitorId) {
    const detail = await wait(monitorId);
    if (detail.status !== STATUS_COMPLETE) return abortRun(payload);
  }

  return startTraining(payload);
}

async Task OnTrainHookAsync(JsonObject payload)
{
    string monitorId = J.Str(payload, "monitorId");

    if (monitorId != null)
    {
        JsonObject detail = await client.WaitAsync(monitorId, 7200);
        if (J.Int(detail, "status", -1) != InnorixClient.StatusComplete)
        {
            await AbortRunAsync(payload);
            return;
        }
    }

    await StartTrainingAsync(payload);
}

runs = api("GET", f"/api/automations/{deploy_id}/executions") or []
latest = runs[0] if runs else None

if latest and latest["status"] != STATUS_COMPLETE:
    print("deploy failed:", latest["monitorId"],
          "retried", retry_failed(latest["monitorId"]), "files")

List<Map<String, Object>> runs = client.executions(deployId);
Map<String, Object> latest = runs.isEmpty() ? null : runs.get(0);

if (latest != null
        && Json.intOr(latest, "status", -1) != InnorixClient.STATUS_COMPLETE) {
    int count = client.retryFailed(Json.str(latest, "monitorId"));
    System.out.println("배포 실패: " + Json.str(latest, "monitorId")
            + " 재전송 " + count + "건");
}

const runs = (await client.executions(deployId)) || [];
const latest = runs[0] || null;

if (latest && latest.status !== STATUS_COMPLETE) {
  const count = await client.retryFailed(latest.monitorId);
  console.log("배포 실패:", latest.monitorId, "재전송", count, "건");
}

List<JsonObject> runs = J.AsList(await client.ExecutionsAsync(deployId));
JsonObject latest = runs.Count > 0 ? runs[0] : null;

if (latest != null
    && J.IntOrNull(latest, "status") != InnorixClient.StatusComplete)
{
    int count = await client.RetryFailedAsync(J.Str(latest, "monitorId"));
    Console.WriteLine(
quot;배포 실패: {J.Str(latest, "monitorId")} 재전송 {count}건");
}