diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs
index 4de435d9..b06d742f 100644
--- a/crates/taskito-java/src/convert.rs
+++ b/crates/taskito-java/src/convert.rs
@@ -7,8 +7,8 @@ use serde::{Deserialize, Serialize};
 use taskito_core::job::{now_millis, Job, NewJob};
 use taskito_core::resilience::circuit_breaker::CircuitState;
 use taskito_core::storage::models::{
-    CircuitBreakerRow, JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow,
-    WorkerRow,
+    CircuitBreakerRow, JobErrorRow, LockInfoRow, PeriodicTaskRow, ReplayHistoryRow, TaskLogRow,
+    TaskMetricRow, WorkerRow,
 };
 use taskito_core::storage::{DeadJob, QueueStats};
 
@@ -379,6 +379,10 @@ pub struct WorkerView<'a> {
     pub pool_type: Option<&'a str>,
     pub threads: i32,
     pub tags: Option<&'a str>,
+    /// JSON array of resource names the worker advertised at registration.
+    pub resources: Option<&'a str>,
+    /// JSON object of per-resource health, written by the worker's heartbeat.
+    pub resource_health: Option<&'a str>,
 }
 
 impl<'a> From<&'a WorkerRow> for WorkerView<'a> {
@@ -394,6 +398,33 @@ impl<'a> From<&'a WorkerRow> for WorkerView<'a> {
             pool_type: w.pool_type.as_deref(),
             threads: w.threads,
             tags: w.tags.as_deref(),
+            resources: w.resources.as_deref(),
+            resource_health: w.resource_health.as_deref(),
+        }
+    }
+}
+
+/// One replay of a job (result blobs are omitted; ids + errors suffice).
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReplayEntryView<'a> {
+    pub id: &'a str,
+    pub original_job_id: &'a str,
+    pub replay_job_id: &'a str,
+    pub replayed_at: i64,
+    pub original_error: Option<&'a str>,
+    pub replay_error: Option<&'a str>,
+}
+
+impl<'a> From<&'a ReplayHistoryRow> for ReplayEntryView<'a> {
+    fn from(r: &'a ReplayHistoryRow) -> Self {
+        Self {
+            id: &r.id,
+            original_job_id: &r.original_job_id,
+            replay_job_id: &r.replay_job_id,
+            replayed_at: r.replayed_at,
+            original_error: r.original_error.as_deref(),
+            replay_error: r.replay_error.as_deref(),
         }
     }
 }
diff --git a/crates/taskito-java/src/queue/admin.rs b/crates/taskito-java/src/queue/admin.rs
index 5aa1f356..ad7d6ffa 100644
--- a/crates/taskito-java/src/queue/admin.rs
+++ b/crates/taskito-java/src/queue/admin.rs
@@ -3,10 +3,12 @@
 use jni::objects::{JClass, JString};
 use jni::sys::{jboolean, jlong, jstring, JNI_FALSE};
 use jni::JNIEnv;
+use taskito_core::job::{now_millis, NewJob};
 use taskito_core::Storage;
 
 use super::borrow_queue;
-use crate::convert::{to_json, DeadJobView};
+use crate::convert::{to_json, DeadJobView, ReplayEntryView};
+use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_string};
 
 /// `String listDead(long handle, long limit, long offset)` — dead-letter entries.
@@ -229,3 +231,67 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listSettin
         new_string(env, to_json(&queue.storage.list_settings()?)?)
     })
 }
+
+/// `String replayJob(long handle, String jobId)` — re-enqueue a copy of an
+/// existing job and record it in the replay history. Returns the new job id.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_replayJob<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let original = queue
+            .storage
+            .get_job(&id)?
+            .ok_or_else(|| BindingError::new(format!("Job not found: {id}")))?;
+        let new_job = NewJob {
+            queue: original.queue.clone(),
+            task_name: original.task_name.clone(),
+            payload: original.payload.clone(),
+            priority: original.priority,
+            scheduled_at: now_millis(),
+            max_retries: original.max_retries,
+            timeout_ms: original.timeout_ms,
+            unique_key: None,
+            metadata: Some(format!("{{\"replayed_from\":\"{id}\"}}")),
+            notes: original.notes.clone(),
+            depends_on: Vec::new(),
+            expires_at: None,
+            result_ttl_ms: original.result_ttl_ms,
+            namespace: original.namespace.clone(),
+        };
+        let job = queue.storage.enqueue(new_job)?;
+        // Best-effort audit row — a history write must not fail the replay.
+        let _ = queue.storage.record_replay(
+            &id,
+            &job.id,
+            original.result.as_deref(),
+            None,
+            original.error.as_deref(),
+            None,
+        );
+        new_string(env, job.id)
+    })
+}
+
+/// `String getReplayHistory(long handle, String jobId)` — replays recorded for
+/// a job, as a JSON array.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getReplayHistory<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let rows = queue.storage.get_replay_history(&id)?;
+        let views: Vec<ReplayEntryView> = rows.iter().map(ReplayEntryView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
diff --git a/crates/taskito-java/src/queue/inspect.rs b/crates/taskito-java/src/queue/inspect.rs
index 8505bbd7..5c958693 100644
--- a/crates/taskito-java/src/queue/inspect.rs
+++ b/crates/taskito-java/src/queue/inspect.rs
@@ -1,10 +1,12 @@
 //! Read-only inspection entry points for `NativeQueue`.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use jni::objects::{JClass, JString};
 use jni::sys::{jlong, jstring};
 use jni::JNIEnv;
+use serde::Serialize;
+use taskito_core::job::Job;
 use taskito_core::Storage;
 
 use super::borrow_queue;
@@ -15,6 +17,22 @@ use crate::convert::{
 use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_optional_string, read_string};
 
+/// One `dependency -> dependent` edge in a job DAG.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct DagEdgeView {
+    from: String,
+    to: String,
+}
+
+/// The dependency DAG reachable from a job: full job rows plus edges.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct JobDagView<'a> {
+    nodes: Vec<JobView<'a>>,
+    edges: Vec<DagEdgeView>,
+}
+
 const DEFAULT_LIMIT: i64 = 50;
 
 /// `String stats(long handle)` — job counts by status across all queues.
@@ -165,3 +183,56 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listCircui
         new_string(env, to_json(&views)?)
     })
 }
+
+/// `String jobDag(long handle, String jobId)` — the dependency DAG reachable
+/// from a job as a JSON `{nodes, edges}` object. Walks dependencies and
+/// dependents in both directions, deduping nodes and edges.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_jobDag<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let start = read_string(env, &job_id)?;
+
+        let mut visited: HashSet<String> = HashSet::new();
+        // Both endpoints of an edge report it; dedupe by (from, to).
+        let mut seen_edges: HashSet<(String, String)> = HashSet::new();
+        // Keep the jobs alive so the borrowing `JobView`s can be built at the end.
+        let mut jobs: Vec<Job> = Vec::new();
+        let mut edges: Vec<DagEdgeView> = Vec::new();
+        let mut pending = vec![start];
+        while let Some(current) = pending.pop() {
+            if !visited.insert(current.clone()) {
+                continue;
+            }
+            let Some(job) = queue.storage.get_job(&current)? else {
+                continue;
+            };
+            jobs.push(job);
+            for dep_id in queue.storage.get_dependencies(&current)? {
+                if seen_edges.insert((dep_id.clone(), current.clone())) {
+                    edges.push(DagEdgeView {
+                        from: dep_id.clone(),
+                        to: current.clone(),
+                    });
+                }
+                pending.push(dep_id);
+            }
+            for dep_id in queue.storage.get_dependents(&current)? {
+                if seen_edges.insert((current.clone(), dep_id.clone())) {
+                    edges.push(DagEdgeView {
+                        from: current.clone(),
+                        to: dep_id.clone(),
+                    });
+                }
+                pending.push(dep_id);
+            }
+        }
+        let nodes: Vec<JobView> = jobs.iter().map(JobView::from).collect();
+        new_string(env, to_json(&JobDagView { nodes, edges })?)
+    })
+}
diff --git a/crates/taskito-java/src/queue/logs.rs b/crates/taskito-java/src/queue/logs.rs
index d6e418f4..284b1b13 100644
--- a/crates/taskito-java/src/queue/logs.rs
+++ b/crates/taskito-java/src/queue/logs.rs
@@ -51,3 +51,51 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getTaskLog
         new_string(env, to_json(&views)?)
     })
 }
+
+/// `String getTaskLogsAfter(long handle, String jobId, String afterIdOrNull)` —
+/// a JSON array of log lines with id strictly after `afterId` (`null` = all).
+/// UUIDv7 ids are time-ordered, so the last id doubles as a stream cursor.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getTaskLogsAfter<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+    after_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let after = read_optional_string(env, &after_id)?;
+        let logs = queue.storage.get_task_logs_after(&id, after.as_deref())?;
+        let views: Vec<LogView> = logs.iter().map(LogView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `String queryTaskLogs(long handle, String taskOrNull, String levelOrNull, long sinceMs, long limit)`
+/// — logs across jobs filtered by task/level, at or after `sinceMs`, newest capped at `limit`.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_queryTaskLogs<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    task_name: JString<'local>,
+    level: JString<'local>,
+    since_ms: jlong,
+    limit: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let task = read_optional_string(env, &task_name)?;
+        let level = read_optional_string(env, &level)?;
+        let logs = queue.storage.query_task_logs(
+            task.as_deref(),
+            level.as_deref(),
+            since_ms,
+            limit.max(0),
+        )?;
+        let views: Vec<LogView> = logs.iter().map(LogView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
index 05f77a1e..ba86ef93 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
@@ -32,10 +32,12 @@
 import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
+import org.byteveda.taskito.model.JobDag;
 import org.byteveda.taskito.model.JobError;
 import org.byteveda.taskito.model.JobFilter;
 import org.byteveda.taskito.model.PeriodicInfo;
 import org.byteveda.taskito.model.QueueStats;
+import org.byteveda.taskito.model.ReplayEntry;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
@@ -506,6 +508,21 @@ public String retry(String deadId) {
         return backend.retryDead(deadId);
     }
 
+    @Override
+    public String replayJob(String jobId) {
+        return backend.replayJob(jobId);
+    }
+
+    @Override
+    public List<ReplayEntry> getReplayHistory(String jobId) {
+        return decodeList(backend.getReplayHistoryJson(jobId), ReplayEntry.class);
+    }
+
+    @Override
+    public JobDag jobDag(String jobId) {
+        return decode(backend.jobDagJson(jobId), JobDag.class);
+    }
+
     @Override
     public boolean deleteDead(String deadId) {
         return backend.deleteDead(deadId);
@@ -573,6 +590,16 @@ public List<TaskLog> getTaskLogs(String jobId) {
         return decodeList(backend.getTaskLogsJson(jobId), TaskLog.class);
     }
 
+    @Override
+    public List<TaskLog> getTaskLogsAfter(String jobId, String afterId) {
+        return decodeList(backend.getTaskLogsAfterJson(jobId, afterId), TaskLog.class);
+    }
+
+    @Override
+    public List<TaskLog> queryTaskLogs(String taskName, String level, long sinceMs, long limit) {
+        return decodeList(backend.queryTaskLogsJson(taskName, level, sinceMs, limit), TaskLog.class);
+    }
+
     @Override
     public Lock lock(String name, long ttlMs) {
         return new Lock(backend, name, ttlMs);
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
index 2642146c..f0929421 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
@@ -24,10 +24,12 @@
 import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
+import org.byteveda.taskito.model.JobDag;
 import org.byteveda.taskito.model.JobError;
 import org.byteveda.taskito.model.JobFilter;
 import org.byteveda.taskito.model.PeriodicInfo;
 import org.byteveda.taskito.model.QueueStats;
+import org.byteveda.taskito.model.ReplayEntry;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
@@ -198,6 +200,15 @@ static Builder builder() {
     /** Alias of {@link #retryDead(String)} in the guide's vocabulary. */
     String retry(String deadId);
 
+    /** Re-enqueue a copy of a job (recording it in the replay history); returns the new job id. */
+    String replayJob(String jobId);
+
+    /** A job's replay history. */
+    List<ReplayEntry> getReplayHistory(String jobId);
+
+    /** The dependency DAG reachable from a job (nodes plus {@code from → to} edges). */
+    JobDag jobDag(String jobId);
+
     boolean deleteDead(String deadId);
 
     long purgeDead(long olderThanMs);
@@ -223,6 +234,12 @@ static Builder builder() {
 
     List<TaskLog> getTaskLogs(String jobId);
 
+    /** Logs for a job with id after {@code afterId} (UUIDv7-ordered cursor); null = all. */
+    List<TaskLog> getTaskLogsAfter(String jobId, String afterId);
+
+    /** Logs across jobs filtered by task/level, at or after {@code sinceMs}, capped at {@code limit}. */
+    List<TaskLog> queryTaskLogs(String taskName, String level, long sinceMs, long limit);
+
     // ── Locks ───────────────────────────────────────────────────────
 
     /** A distributed lock {@code name} with the given TTL; call {@link Lock#acquire()}. */
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java
index c2aa5112..27f7621b 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java
@@ -225,8 +225,12 @@ private Router buildRouter() {
         r.get("/api/stats/queues", req -> core.statsByQueue());
         r.get("/api/queues/paused", req -> core.queuesPaused());
         r.get("/api/jobs", req -> core.listJobs(req.query()));
+        r.get("/api/jobs/([^/]+)/logs", req -> core.jobLogs(req.param(0)));
+        r.get("/api/jobs/([^/]+)/replay-history", req -> core.replayHistory(req.param(0)));
+        r.get("/api/jobs/([^/]+)/dag", req -> core.jobDag(req.param(0)));
         r.get("/api/jobs/([^/]+)", req -> core.job(req.param(0)));
         r.get("/api/dead-letters", req -> core.listDead(req.query()));
+        r.get("/api/logs", req -> core.logs(req.query()));
         r.get("/api/workers", req -> core.listWorkers());
 
         // Metrics (aggregated — the SPA contract, not raw rows)
@@ -237,6 +241,7 @@ private Router buildRouter() {
         r.get("/api/circuit-breakers", req -> ops.circuitBreakers());
         r.get("/api/event-types", req -> ops.eventTypes());
         r.get("/api/scaler", req -> ops.scaler(req.query()));
+        r.get("/api/resources", req -> ops.resources());
 
         // Settings KV
         r.get("/api/settings", req -> settingsApi.list());
@@ -255,6 +260,7 @@ private Router buildRouter() {
         r.delete("/api/queues/([^/]+)/override", req -> overrides.deleteQueueOverride(req.param(0)));
 
         // Action
+        r.post("/api/jobs/([^/]+)/replay", req -> core.replayJob(req.param(0)));
         r.post("/api/jobs/([^/]+)/cancel", req -> core.cancel(req.param(0)));
         r.post("/api/dead-letters/([^/]+)/retry", req -> core.retryDead(req.param(0)));
         r.post("/api/queues/([^/]+)/pause", req -> core.pause(req.param(0)));
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java
index 985e77e5..d5e0c36d 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java
@@ -2,10 +2,15 @@
 
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
 import org.byteveda.taskito.model.CircuitBreakerState;
+import org.byteveda.taskito.model.DagEdge;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
+import org.byteveda.taskito.model.JobDag;
 import org.byteveda.taskito.model.QueueStats;
+import org.byteveda.taskito.model.ReplayEntry;
+import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.WorkerInfo;
 
 /**
@@ -91,4 +96,47 @@ static Map<String, Object> dead(DeadJob d) {
         m.put("dlq_retry_count", d.dlqRetryCount);
         return m;
     }
+
+    static Map<String, Object> taskLog(TaskLog l) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("id", l.id);
+        m.put("job_id", l.jobId);
+        m.put("task_name", l.taskName);
+        m.put("level", l.level);
+        m.put("message", l.message);
+        m.put("extra", l.extra);
+        m.put("logged_at", l.loggedAt);
+        return m;
+    }
+
+    static Map<String, Object> replayEntry(ReplayEntry r) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("replay_job_id", r.replayJobId);
+        m.put("replayed_at", r.replayedAt);
+        m.put("original_error", r.originalError);
+        m.put("replay_error", r.replayError);
+        return m;
+    }
+
+    static Map<String, Object> jobDag(JobDag dag) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("nodes", dag.nodes.stream().map(Contract::dagNode).collect(Collectors.toList()));
+        m.put("edges", dag.edges.stream().map(Contract::dagEdge).collect(Collectors.toList()));
+        return m;
+    }
+
+    private static Map<String, Object> dagNode(Job j) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("id", j.id);
+        m.put("task_name", j.taskName);
+        m.put("status", j.status.wire());
+        return m;
+    }
+
+    private static Map<String, Object> dagEdge(DagEdge e) {
+        Map<String, Object> m = new LinkedHashMap<>();
+        m.put("from", e.from);
+        m.put("to", e.to);
+        return m;
+    }
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java
index 2980211a..d547af0b 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/CoreHandlers.java
@@ -87,4 +87,30 @@ public Object resume(String name) {
         queue.queue(name).resume();
         return Map.of("ok", true);
     }
+
+    /** Logs across jobs filtered by task/level; {@code since} is a lookback in seconds. */
+    public Object logs(Map<String, String> query) {
+        long sinceSeconds = Http.longParam(query, "since", 3600);
+        long sinceMs = System.currentTimeMillis() - sinceSeconds * 1000;
+        long limit = Http.longParam(query, "limit", 100);
+        return queue.queryTaskLogs(query.get("task"), query.get("level"), sinceMs, limit).stream()
+                .map(Contract::taskLog)
+                .collect(Collectors.toList());
+    }
+
+    public Object jobLogs(String id) {
+        return queue.getTaskLogs(id).stream().map(Contract::taskLog).collect(Collectors.toList());
+    }
+
+    public Object replayJob(String id) {
+        return Map.of("replay_job_id", queue.replayJob(id));
+    }
+
+    public Object replayHistory(String id) {
+        return queue.getReplayHistory(id).stream().map(Contract::replayEntry).collect(Collectors.toList());
+    }
+
+    public Object jobDag(String id) {
+        return Contract.jobDag(queue.jobDag(id));
+    }
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java
index 7e593cc3..c6eb44e4 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/OpsHandlers.java
@@ -5,9 +5,13 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
 import java.util.stream.Collectors;
 import org.byteveda.taskito.Taskito;
 import org.byteveda.taskito.dashboard.support.Http;
+import org.byteveda.taskito.dashboard.support.Json;
 import org.byteveda.taskito.events.EventName;
 import org.byteveda.taskito.model.QueueStats;
 import org.byteveda.taskito.model.WorkerInfo;
@@ -62,6 +66,54 @@ public Object scaler(Map<String, String> query) {
         return out;
     }
 
+    /**
+     * Per-resource status aggregated across workers, derived from each worker's
+     * advertised {@code resources} + reported {@code resourceHealth}. A resource
+     * a worker advertises but never reports on is {@code not_initialized}; when
+     * workers disagree, the worst health wins.
+     */
+    public Object resources() {
+        Map<String, Integer> reported = new TreeMap<>();
+        Set<String> advertised = new TreeSet<>();
+        for (WorkerInfo worker : queue.listWorkers()) {
+            advertised.addAll(Json.parseStringList(worker.resources));
+            Map<String, Object> health = Json.parseMap(worker.resourceHealth);
+            if (health != null) {
+                health.forEach((name, value) -> reported.merge(name, severity(String.valueOf(value)), Math::max));
+            }
+        }
+        Set<String> all = new TreeSet<>(advertised);
+        all.addAll(reported.keySet());
+        List<Map<String, Object>> out = new ArrayList<>();
+        for (String name : all) {
+            Map<String, Object> entry = new LinkedHashMap<>();
+            entry.put("name", name);
+            entry.put("scope", "worker");
+            entry.put("health", reported.containsKey(name) ? healthName(reported.get(name)) : "not_initialized");
+            entry.put("init_duration_ms", 0);
+            entry.put("recreations", 0);
+            entry.put("depends_on", List.of());
+            out.add(entry);
+        }
+        return out;
+    }
+
+    private static int severity(String health) {
+        return switch (health) {
+            case "unhealthy" -> 2;
+            case "degraded" -> 1;
+            default -> 0;
+        };
+    }
+
+    private static String healthName(int severity) {
+        return switch (severity) {
+            case 2 -> "unhealthy";
+            case 1 -> "degraded";
+            default -> "healthy";
+        };
+    }
+
     public Object readiness() {
         Map<String, Object> checks = new LinkedHashMap<>();
         checks.put("storage", true);
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Json.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Json.java
index fa159b66..8ece2ca1 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Json.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/support/Json.java
@@ -4,6 +4,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import org.byteveda.taskito.errors.SerializationException;
 
@@ -58,6 +60,24 @@ public static Map<String, Object> parseMap(String json) {
         }
     }
 
+    /** Parse a JSON array of strings; empty list if malformed/non-array/null. */
+    public static List<String> parseStringList(String json) {
+        if (json == null || json.isEmpty()) {
+            return List.of();
+        }
+        try {
+            JsonNode node = MAPPER.readTree(json);
+            if (node == null || !node.isArray()) {
+                return List.of();
+            }
+            List<String> out = new ArrayList<>(node.size());
+            node.forEach(element -> out.add(element.asText()));
+            return out;
+        } catch (IOException e) {
+            return List.of();
+        }
+    }
+
     private static Map<String, Object> asMap(JsonNode node) {
         if (node == null || !node.isObject()) {
             return null;
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
index 73a982b4..07cf43de 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
@@ -137,6 +137,21 @@ public String retryDead(String deadId) {
         return withOpenHandle(() -> NativeQueue.retryDead(handle, deadId));
     }
 
+    @Override
+    public String replayJob(String jobId) {
+        return withOpenHandle(() -> NativeQueue.replayJob(handle, jobId));
+    }
+
+    @Override
+    public String getReplayHistoryJson(String jobId) {
+        return withOpenHandle(() -> NativeQueue.getReplayHistory(handle, jobId));
+    }
+
+    @Override
+    public String jobDagJson(String jobId) {
+        return withOpenHandle(() -> NativeQueue.jobDag(handle, jobId));
+    }
+
     @Override
     public boolean deleteDead(String deadId) {
         return withOpenHandle(() -> NativeQueue.deleteDead(handle, deadId));
@@ -219,6 +234,16 @@ public String getTaskLogsJson(String jobId) {
         return withOpenHandle(() -> NativeQueue.getTaskLogs(handle, jobId));
     }
 
+    @Override
+    public String getTaskLogsAfterJson(String jobId, String afterIdOrNull) {
+        return withOpenHandle(() -> NativeQueue.getTaskLogsAfter(handle, jobId, afterIdOrNull));
+    }
+
+    @Override
+    public String queryTaskLogsJson(String taskNameOrNull, String levelOrNull, long sinceMs, long limit) {
+        return withOpenHandle(() -> NativeQueue.queryTaskLogs(handle, taskNameOrNull, levelOrNull, sinceMs, limit));
+    }
+
     @Override
     public boolean acquireLock(String name, String ownerId, long ttlMs) {
         return withOpenHandle(() -> NativeQueue.acquireLock(handle, name, ownerId, ttlMs));
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
index 715026bd..c84bb6d7 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
@@ -55,6 +55,12 @@ private NativeQueue() {}
 
     public static native String listCircuitBreakers(long handle);
 
+    public static native String replayJob(long handle, String jobId);
+
+    public static native String getReplayHistory(long handle, String jobId);
+
+    public static native String jobDag(long handle, String jobId);
+
     // ── Admin ───────────────────────────────────────────────────────
     public static native String listDead(long handle, long limit, long offset);
 
@@ -93,6 +99,11 @@ public static native void writeTaskLog(
 
     public static native String getTaskLogs(long handle, String jobId);
 
+    public static native String getTaskLogsAfter(long handle, String jobId, String afterIdOrNull);
+
+    public static native String queryTaskLogs(
+            long handle, String taskNameOrNull, String levelOrNull, long sinceMs, long limit);
+
     // ── Locks ───────────────────────────────────────────────────────
     public static native boolean acquireLock(long handle, String name, String ownerId, long ttlMs);
 
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/DagEdge.java b/sdks/java/src/main/java/org/byteveda/taskito/model/DagEdge.java
new file mode 100644
index 00000000..699b6b42
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/model/DagEdge.java
@@ -0,0 +1,18 @@
+package org.byteveda.taskito.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/** A directed {@code from → to} dependency edge in a job DAG. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public final class DagEdge {
+    public final String from;
+    public final String to;
+
+    @JsonCreator
+    public DagEdge(@JsonProperty("from") String from, @JsonProperty("to") String to) {
+        this.from = from;
+        this.to = to;
+    }
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/JobDag.java b/sdks/java/src/main/java/org/byteveda/taskito/model/JobDag.java
new file mode 100644
index 00000000..f97478bc
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/model/JobDag.java
@@ -0,0 +1,19 @@
+package org.byteveda.taskito.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+
+/** A job's dependency graph: full job rows as nodes plus directed edges. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public final class JobDag {
+    public final List<Job> nodes;
+    public final List<DagEdge> edges;
+
+    @JsonCreator
+    public JobDag(@JsonProperty("nodes") List<Job> nodes, @JsonProperty("edges") List<DagEdge> edges) {
+        this.nodes = nodes == null ? List.of() : List.copyOf(nodes);
+        this.edges = edges == null ? List.of() : List.copyOf(edges);
+    }
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/ReplayEntry.java b/sdks/java/src/main/java/org/byteveda/taskito/model/ReplayEntry.java
new file mode 100644
index 00000000..eba51bcc
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/model/ReplayEntry.java
@@ -0,0 +1,32 @@
+package org.byteveda.taskito.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/** One entry in a job's replay history. {@code replayedAt} is Unix milliseconds. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public final class ReplayEntry {
+    public final String id;
+    public final String originalJobId;
+    public final String replayJobId;
+    public final long replayedAt;
+    public final String originalError;
+    public final String replayError;
+
+    @JsonCreator
+    public ReplayEntry(
+            @JsonProperty("id") String id,
+            @JsonProperty("originalJobId") String originalJobId,
+            @JsonProperty("replayJobId") String replayJobId,
+            @JsonProperty("replayedAt") long replayedAt,
+            @JsonProperty("originalError") String originalError,
+            @JsonProperty("replayError") String replayError) {
+        this.id = id;
+        this.originalJobId = originalJobId;
+        this.replayJobId = replayJobId;
+        this.replayedAt = replayedAt;
+        this.originalError = originalError;
+        this.replayError = replayError;
+    }
+}
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java
index 2088291a..8c19ef6b 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java
@@ -17,6 +17,10 @@ public final class WorkerInfo {
     public final String poolType;
     public final int threads;
     public final String tags;
+    /** JSON array of resource names the worker advertised at registration; may be null. */
+    public final String resources;
+    /** JSON object of per-resource health written by the worker's heartbeat; may be null. */
+    public final String resourceHealth;
 
     @JsonCreator
     public WorkerInfo(
@@ -29,7 +33,9 @@ public WorkerInfo(
             @JsonProperty("pid") Integer pid,
             @JsonProperty("poolType") String poolType,
             @JsonProperty("threads") int threads,
-            @JsonProperty("tags") String tags) {
+            @JsonProperty("tags") String tags,
+            @JsonProperty("resources") String resources,
+            @JsonProperty("resourceHealth") String resourceHealth) {
         this.workerId = workerId;
         this.queues = queues;
         this.status = status;
@@ -40,5 +46,7 @@ public WorkerInfo(
         this.poolType = poolType;
         this.threads = threads;
         this.tags = tags;
+        this.resources = resources;
+        this.resourceHealth = resourceHealth;
     }
 }
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
index 69f054ef..63c55a9a 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
@@ -62,6 +62,21 @@ default String listCircuitBreakersJson() {
 
     String retryDead(String deadId);
 
+    /** Re-enqueue a copy of a job and return the new id. */
+    default String replayJob(String jobId) {
+        throw new UnsupportedOperationException("replay not supported by this backend");
+    }
+
+    /** A job's replay history; defaults to none. */
+    default String getReplayHistoryJson(String jobId) {
+        return "[]";
+    }
+
+    /** A job's dependency DAG; defaults to just the empty graph. */
+    default String jobDagJson(String jobId) {
+        return "{\"nodes\":[],\"edges\":[]}";
+    }
+
     boolean deleteDead(String deadId);
 
     long purgeDead(long olderThanMs);
@@ -97,6 +112,16 @@ default long purgeDeadByTask(String taskName) {
 
     String getTaskLogsJson(String jobId);
 
+    /** Logs for a job after a cursor id; defaults to none for backends without cursor support. */
+    default String getTaskLogsAfterJson(String jobId, String afterIdOrNull) {
+        return "[]";
+    }
+
+    /** Logs across jobs filtered by task/level/since; defaults to none. */
+    default String queryTaskLogsJson(String taskNameOrNull, String levelOrNull, long sinceMs, long limit) {
+        return "[]";
+    }
+
     // ── Locks ───────────────────────────────────────────────────────
     // Optional capability: default to throwing so existing custom backends keep
     // compiling and fail explicitly only when locks are actually used.
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardNativeTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardNativeTest.java
new file mode 100644
index 00000000..1336a278
--- /dev/null
+++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardNativeTest.java
@@ -0,0 +1,87 @@
+package org.byteveda.taskito.dashboard;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Map;
+import org.byteveda.taskito.Taskito;
+import org.byteveda.taskito.model.JobFilter;
+import org.byteveda.taskito.task.EnqueueOptions;
+import org.byteveda.taskito.task.Task;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Coverage of the native-backed endpoints: logs, replay, job DAG, resources. */
+@Timeout(30)
+class DashboardNativeTest {
+
+    @SuppressWarnings("unchecked")
+    private static Class<Map<String, Object>> mapType() {
+        return (Class<Map<String, Object>>) (Class<?>) Map.class;
+    }
+
+    private static String seedJob(Taskito queue) {
+        Task<Map<String, Object>> task = Task.of("email.send", mapType());
+        queue.enqueue(
+                task,
+                Collections.singletonMap("a", 1),
+                EnqueueOptions.builder().queue("emails").build());
+        return queue.listJobs(JobFilter.builder().limit(1).build()).get(0).id;
+    }
+
+    @Test
+    void logsAndJobLogs(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                        Taskito.builder().sqlite(dir.resolve("t.db").toString()).open();
+                DashboardServer server = DashboardServer.start(queue, 0)) {
+            String jobId = seedJob(queue);
+            queue.writeTaskLog(jobId, "email.send", "info", "hello-log");
+            DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue));
+            assertTrue(client.get("/api/jobs/" + jobId + "/logs").body().contains("hello-log"));
+            assertTrue(client.get("/api/logs").body().contains("hello-log"));
+        }
+    }
+
+    @Test
+    void replayAndHistory(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                        Taskito.builder().sqlite(dir.resolve("t.db").toString()).open();
+                DashboardServer server = DashboardServer.start(queue, 0)) {
+            String jobId = seedJob(queue);
+            DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue));
+            HttpResponse<String> replay = client.post("/api/jobs/" + jobId + "/replay", null);
+            assertEquals(200, replay.statusCode());
+            assertTrue(replay.body().contains("replay_job_id"));
+            assertTrue(
+                    client.get("/api/jobs/" + jobId + "/replay-history").body().contains("replay_job_id"));
+        }
+    }
+
+    @Test
+    void jobDagContainsTheJob(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                        Taskito.builder().sqlite(dir.resolve("t.db").toString()).open();
+                DashboardServer server = DashboardServer.start(queue, 0)) {
+            String jobId = seedJob(queue);
+            DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue));
+            String dag = client.get("/api/jobs/" + jobId + "/dag").body();
+            assertTrue(dag.contains("\"nodes\"") && dag.contains("\"edges\""));
+            assertTrue(dag.contains(jobId));
+        }
+    }
+
+    @Test
+    void resourcesEndpointServed(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                        Taskito.builder().sqlite(dir.resolve("t.db").toString()).open();
+                DashboardServer server = DashboardServer.start(queue, 0)) {
+            DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue));
+            // No workers running → an empty resource list.
+            assertEquals("[]", client.get("/api/resources").body());
+        }
+    }
+}