feat(java): workflow dashboard + saga node fields

crates/taskito-java/src/workflows/mod.rs

@@ -97,6 +97,10 @@ struct NodeView<'a> {
     started_at: Option<i64>,
     completed_at: Option<i64>,
     error: Option<&'a str>,
+    compensation_job_id: Option<&'a str>,
+    compensation_started_at: Option<i64>,
+    compensation_completed_at: Option<i64>,
+    compensation_error: Option<&'a str>,
 }
 
 impl<'a> From<&'a WorkflowNode> for NodeView<'a> {
@@ -110,6 +114,43 @@ impl<'a> From<&'a WorkflowNode> for NodeView<'a> {
             started_at: n.started_at,
             completed_at: n.completed_at,
             error: n.error.as_deref(),
+            compensation_job_id: n.compensation_job_id.as_deref(),
+            compensation_started_at: n.compensation_started_at,
+            compensation_completed_at: n.compensation_completed_at,
+            compensation_error: n.compensation_error.as_deref(),
+        }
+    }
+}
+
+/// A workflow run summary (no node detail — `getWorkflowStatus` provides that).
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct WorkflowRunView<'a> {
+    id: &'a str,
+    definition_id: &'a str,
+    state: &'static str,
+    params: Option<&'a str>,
+    error: Option<&'a str>,
+    started_at: Option<i64>,
+    completed_at: Option<i64>,
+    created_at: i64,
+    parent_run_id: Option<&'a str>,
+    parent_node_name: Option<&'a str>,
+}
+
+impl<'a> From<&'a WorkflowRun> for WorkflowRunView<'a> {
+    fn from(r: &'a WorkflowRun) -> Self {
+        Self {
+            id: &r.id,
+            definition_id: &r.definition_id,
+            state: r.state.as_str(),
+            params: r.params.as_deref(),
+            error: r.error.as_deref(),
+            started_at: r.started_at,
+            completed_at: r.completed_at,
+            created_at: r.created_at,
+            parent_run_id: r.parent_run_id.as_deref(),
+            parent_node_name: r.parent_node_name.as_deref(),
         }
     }
 }
@@ -1348,3 +1389,107 @@ fn cascade_skip_pending(
         }
     }
 }
+
+/// `String listWorkflowRuns(long, String definitionNameOrNull, String stateOrNull,
+/// long limit, long offset)` — run summaries as a JSON array (dashboard read).
+#[no_mangle]
+#[allow(clippy::too_many_arguments)]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_listWorkflowRuns<
+    'local,
+>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    definition_name: JString<'local>,
+    state: JString<'local>,
+    limit: jlong,
+    offset: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let definition_name = read_optional_string(env, &definition_name)?;
+        let state_raw = read_optional_string(env, &state)?;
+        let state = match state_raw.as_deref() {
+            Some(s) => Some(
+                WorkflowState::from_str_val(s)
+                    .ok_or_else(|| BindingError::new(format!("unknown workflow state '{s}'")))?,
+            ),
+            None => None,
+        };
+        let wf = queue.workflow_store()?;
+        let runs = wf.list_workflow_runs(
+            definition_name.as_deref(),
+            state,
+            limit.max(0),
+            offset.max(0),
+        )?;
+        let views: Vec<WorkflowRunView> = runs.iter().map(WorkflowRunView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `String getWorkflowRun(long, String runId)` — a single run summary, or
+/// `null` if the run no longer exists.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowRun<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    run_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let run_id = read_string(env, &run_id)?;
+        let wf = queue.workflow_store()?;
+        match wf.get_workflow_run(&run_id)? {
+            Some(run) => new_string(env, to_json(&WorkflowRunView::from(&run))?),
+            None => Ok(std::ptr::null_mut()),
+        }
+    })
+}
+
+/// `String getWorkflowChildren(long, String runId)` — sub-workflow runs spawned
+/// by a run, as a JSON array of run summaries.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowChildren<
+    'local,
+>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    run_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let run_id = read_string(env, &run_id)?;
+        let wf = queue.workflow_store()?;
+        let runs = wf.get_child_workflow_runs(&run_id)?;
+        let views: Vec<WorkflowRunView> = runs.iter().map(WorkflowRunView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `String getWorkflowDag(long, String runId)` — the serialized DAG JSON backing
+/// a run, or `null` if the run or its definition no longer exists.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowDag<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    run_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let run_id = read_string(env, &run_id)?;
+        let wf = queue.workflow_store()?;
+        let Some(run) = wf.get_workflow_run(&run_id)? else {
+            return Ok(std::ptr::null_mut());
+        };
+        let Some(def) = wf.get_workflow_definition_by_id(&run.definition_id)? else {
+            return Ok(std::ptr::null_mut());
+        };
+        let dag = String::from_utf8(def.dag_data)
+            .map_err(|e| BindingError::new(format!("invalid DAG utf8: {e}")))?;
+        new_string(env, dag)
+    })
+}

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -41,6 +41,7 @@
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
+import org.byteveda.taskito.model.WorkflowRunInfo;
 import org.byteveda.taskito.predicates.EnqueueDecision;
 import org.byteveda.taskito.predicates.EnqueueGate;
 import org.byteveda.taskito.predicates.Predicate;
@@ -774,6 +775,26 @@ public void cancelWorkflow(String runId) {
         backend.cancelWorkflowRun(runId);
     }
 
+    @Override
+    public List<WorkflowRunInfo> listWorkflowRuns(String definitionName, String state, long limit, long offset) {
+        return decodeList(backend.listWorkflowRunsJson(definitionName, state, limit, offset), WorkflowRunInfo.class);
+    }
+
+    @Override
+    public Optional<WorkflowRunInfo> getWorkflowRun(String runId) {
+        return backend.getWorkflowRunJson(runId).map(json -> decode(json, WorkflowRunInfo.class));
+    }
+
+    @Override
+    public List<WorkflowRunInfo> getWorkflowChildren(String runId) {
+        return decodeList(backend.getWorkflowChildrenJson(runId), WorkflowRunInfo.class);
+    }
+
+    @Override
+    public Optional<String> getWorkflowDag(String runId) {
+        return backend.getWorkflowDagJson(runId);
+    }
+
     /** Encode a step's structure + per-step overrides for the native submit call. */
     private Map<String, Object> stepSpec(Step step) {
         Map<String, Object> spec = new LinkedHashMap<>();

sdks/java/src/main/java/org/byteveda/taskito/Taskito.java

@@ -33,6 +33,7 @@
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
+import org.byteveda.taskito.model.WorkflowRunInfo;
 import org.byteveda.taskito.predicates.EnqueueGate;
 import org.byteveda.taskito.predicates.Predicate;
 import org.byteveda.taskito.resources.PoolConfig;
@@ -292,6 +293,18 @@ static Builder builder() {
     /** Cancel a workflow run: skip its pending nodes and mark it cancelled. */
     void cancelWorkflow(String runId);
 
+    /** Workflow run summaries, filtered by definition name and/or state, paged. Nulls mean no filter. */
+    List<WorkflowRunInfo> listWorkflowRuns(String definitionName, String state, long limit, long offset);
+
+    /** A single workflow run summary, or empty if the run no longer exists. */
+    Optional<WorkflowRunInfo> getWorkflowRun(String runId);
+
+    /** Sub-workflow runs spawned by a run. */
+    List<WorkflowRunInfo> getWorkflowChildren(String runId);
+
+    /** The serialized DAG JSON backing a run, or empty if the run/definition is gone. */
+    Optional<String> getWorkflowDag(String runId);
+
     // ── Worker ──────────────────────────────────────────────────────
 
     /** Begin building a worker over this client. */

sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java

@@ -19,6 +19,7 @@
 import org.byteveda.taskito.dashboard.api.OpsHandlers;
 import org.byteveda.taskito.dashboard.api.OverridesHandlers;
 import org.byteveda.taskito.dashboard.api.SettingsHandlers;
+import org.byteveda.taskito.dashboard.api.WorkflowsHandlers;
 import org.byteveda.taskito.dashboard.auth.AuthHandlers;
 import org.byteveda.taskito.dashboard.auth.AuthStore;
 import org.byteveda.taskito.dashboard.auth.Cookies;
@@ -202,6 +203,7 @@ private Router buildRouter() {
         SettingsHandlers settingsApi = new SettingsHandlers(settings);
         OverridesHandlers overrides = new OverridesHandlers(queue, new OverridesStore(settings));
         MetricsHandlers metrics = new MetricsHandlers(queue);
+        WorkflowsHandlers workflows = new WorkflowsHandlers(queue);
         long ttl = AuthStore.DEFAULT_SESSION_TTL_SECONDS;
         Router r = new Router();
 
@@ -243,6 +245,12 @@ private Router buildRouter() {
         r.get("/api/scaler", req -> ops.scaler(req.query()));
         r.get("/api/resources", req -> ops.resources());
 
+        // Workflows (read)
+        r.get("/api/workflows/runs", req -> workflows.runs(req.query()));
+        r.get("/api/workflows/runs/([^/]+)/dag", req -> workflows.dag(req.param(0)));
+        r.get("/api/workflows/runs/([^/]+)/children", req -> workflows.children(req.param(0)));
+        r.get("/api/workflows/runs/([^/]+)", req -> workflows.run(req.param(0)));
+
         // Settings KV
         r.get("/api/settings", req -> settingsApi.list());
         r.get("/api/settings/(.+)", req -> settingsApi.get(req.param(0)));

sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java

@@ -12,6 +12,8 @@
 import org.byteveda.taskito.model.ReplayEntry;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.WorkerInfo;
+import org.byteveda.taskito.model.WorkflowRunInfo;
+import org.byteveda.taskito.workflows.NodeSnapshot;
 
 /**
  * Maps the SDK's camelCase types to the snake_case JSON contract the React SPA
@@ -139,4 +141,36 @@ private static Map<String, Object> dagEdge(DagEdge e) {
         m.put("to", e.to);
         return m;
     }
+
+    static Map<String, Object> workflowRun(WorkflowRunInfo r) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("id", r.id);
+        m.put("definition_id", r.definitionId);
+        m.put("state", r.state);
+        m.put("params", r.params);
+        m.put("started_at", r.startedAt);
+        m.put("completed_at", r.completedAt);
+        m.put("error", r.error);
+        m.put("parent_run_id", r.parentRunId);
+        m.put("parent_node_name", r.parentNodeName);
+        m.put("created_at", r.createdAt);
+        return m;
+    }
+
+    static Map<String, Object> workflowNode(NodeSnapshot n) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("node_name", n.nodeName);
+        m.put("status", n.status);
+        m.put("job_id", n.jobId);
+        m.put("result_hash", n.resultHash);
+        m.put("fan_out_count", n.fanOutCount);
+        m.put("started_at", n.startedAt);
+        m.put("completed_at", n.completedAt);
+        m.put("error", n.error);
+        m.put("compensation_job_id", n.compensationJobId);
+        m.put("compensation_started_at", n.compensationStartedAt);
+        m.put("compensation_completed_at", n.compensationCompletedAt);
+        m.put("compensation_error", n.compensationError);
+        return m;
+    }
 }