From 4fad5c7c48659a99c6ed927c9e27a312571c6b49 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:21:33 +0530 Subject: [PATCH 1/8] docs: specify structured task errors in binding contract --- crates/taskito-core/BINDING_CONTRACT.md | 30 +++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 4dcce751..87c8794e 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/crates/taskito-core/BINDING_CONTRACT.md @@ -96,6 +96,36 @@ Scheduler.handle_result ─▶ ResultOutcome ─▶ shell emits events / middlew Channels: inbound `tokio::sync::mpsc::Receiver` (async); outbound `crossbeam_channel::Sender` (sync, cloneable). +## Task errors (structured, cross-SDK) +When a task raises, the shell reports the failure as a **canonical JSON object** +serialized into `JobResult::Failure.error` (and thus into `jobs.error`, +`job_errors.error`, `dead_letter.error` — the storage layer never interprets it): + +```json +{"errtype": "ValueError", "message": "bad value 42", "traceback": ["...frame...", "..."]} +``` + +- `errtype` — the exception's class name, as idiomatic per language (qualified + where the language has a notion of it). Required. +- `message` — the human-readable message, verbatim (keeps `error_like` + substring filters useful). Required, may be empty. +- `traceback` — array of strings, best-effort per shell; `[]` when the + language/runtime can't provide frames. Required key. + +**Fallback rule (readers)**: an error string that does not parse as a JSON +object with a `message` key is a plain legacy/system string and MUST be +surfaced as-is. Core-generated maintenance errors (timeouts, worker-death +recovery, expiry, cancellation) remain plain strings by design. + +**Retry semantics**: `retry_on`/`dont_retry_on`-style filtering matches on the +live exception object before formatting — the stored string never drives retry +decisions. + +**Test vector** (assert byte-exact in each shell's formatter): +input errtype `BoomError`, message `it broke`, traceback `["frame1", "frame2"]` → +`{"errtype":"BoomError","message":"it broke","traceback":["frame1","frame2"]}` +(JSON with those three keys in that order, no extra whitespace). + ## Types the shell produces / consumes - **`Job`** — `job.rs`. Fields incl. `id`, `queue`, `task_name`, `payload: Vec` (opaque), `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`, From 3033a73eb9e6df017596909e7477e78bb27f3313 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:21:33 +0530 Subject: [PATCH 2/8] feat(python): emit and decode structured task errors --- .../src/native_async/task_executor.rs | 15 +- crates/taskito-python/src/py_worker.rs | 14 +- sdks/python/taskito/async_support/executor.py | 4 +- sdks/python/taskito/exceptions.py | 31 ++++- sdks/python/taskito/prefork/child.py | 8 +- sdks/python/taskito/result.py | 52 +++++-- sdks/python/taskito/task_errors.py | 70 ++++++++++ sdks/python/tests/core/test_task_errors.py | 128 ++++++++++++++++++ 8 files changed, 288 insertions(+), 34 deletions(-) create mode 100644 sdks/python/taskito/task_errors.py create mode 100644 sdks/python/tests/core/test_task_errors.py diff --git a/crates/taskito-python/src/native_async/task_executor.rs b/crates/taskito-python/src/native_async/task_executor.rs index e57cb8a7..1b9c31e8 100644 --- a/crates/taskito-python/src/native_async/task_executor.rs +++ b/crates/taskito-python/src/native_async/task_executor.rs @@ -165,19 +165,8 @@ fn run_task(py: Python<'_>, task_registry: &Py, job: &Job) -> PyResult, e: &PyErr) -> String { - if let Ok(tb_mod) = py.import("traceback") { - if let Ok(formatted) = tb_mod.call_method1( - "format_exception", - (e.get_type(py), e.value(py), e.traceback(py)), - ) { - if let Ok(lines) = formatted.extract::>() { - return lines.join(""); - } - } - } - format!("{e}") -} +// One structured-error encoder for every worker path — see py_worker.rs. +use crate::py_worker::format_python_error; fn is_cancelled_error(py: Python<'_>, e: &PyErr) -> bool { if let Ok(exceptions_mod) = py.import("taskito.exceptions") { diff --git a/crates/taskito-python/src/py_worker.rs b/crates/taskito-python/src/py_worker.rs index 37fde72d..c338cc3d 100644 --- a/crates/taskito-python/src/py_worker.rs +++ b/crates/taskito-python/src/py_worker.rs @@ -224,7 +224,19 @@ pub fn execute_task( } pub fn format_python_error(py: Python<'_>, e: &PyErr) -> String { - // Try to get a full traceback + // Canonical structured error (BINDING_CONTRACT.md "Task errors") — the + // Python module owns the encoding so every worker path emits one format. + if let Ok(errors_mod) = py.import("taskito.task_errors") { + if let Ok(encoded) = errors_mod.call_method1( + "encode_from_parts", + (e.get_type(py), e.value(py), e.traceback(py)), + ) { + if let Ok(json) = encoded.extract::() { + return json; + } + } + } + // Fallback: plain traceback text (readers treat non-JSON as legacy). if let Ok(tb_mod) = py.import("traceback") { if let Ok(formatted) = tb_mod.call_method1( "format_exception", diff --git a/sdks/python/taskito/async_support/executor.py b/sdks/python/taskito/async_support/executor.py index 5685ead4..e9806b45 100644 --- a/sdks/python/taskito/async_support/executor.py +++ b/sdks/python/taskito/async_support/executor.py @@ -6,7 +6,6 @@ import logging import threading import time -import traceback from typing import TYPE_CHECKING, Any from taskito.async_support.context import clear_async_context, set_async_context @@ -14,6 +13,7 @@ from taskito.exceptions import TaskCancelledError from taskito.interception.reconstruct import reconstruct_args from taskito.proxies import cleanup_proxies, reconstruct_proxies +from taskito.task_errors import encode_task_error if TYPE_CHECKING: from taskito.app import Queue @@ -171,7 +171,7 @@ async def _execute( except Exception as exc: error = exc wall_ns = time.monotonic_ns() - start_ns - error_msg = traceback.format_exc() + error_msg = encode_task_error(exc) should_retry = self._check_retry(task_name, exc) self._sender.report_failure( job_id, diff --git a/sdks/python/taskito/exceptions.py b/sdks/python/taskito/exceptions.py index bc386aad..ded8a154 100644 --- a/sdks/python/taskito/exceptions.py +++ b/sdks/python/taskito/exceptions.py @@ -17,11 +17,38 @@ class TaskCancelledError(TaskitoError): """Raised when a running task detects it has been cancelled.""" -class TaskFailedError(TaskitoError): +class JobErrorDetails: + """Structured failure details mixin for job-outcome exceptions. + + Populated when the stored error is a canonical structured task error + (see ``taskito.task_errors``); all fields default to ``None`` for + legacy/system plain-string errors. + """ + + errtype: str | None = None + traceback: list[str] | None = None + job_id: str | None = None + raw_error: str | None = None + + def _attach_details( + self, + *, + errtype: str | None, + traceback: list[str] | None, + job_id: str | None, + raw_error: str | None, + ) -> None: + self.errtype = errtype + self.traceback = traceback + self.job_id = job_id + self.raw_error = raw_error + + +class TaskFailedError(TaskitoError, JobErrorDetails): """Raised when a task has failed.""" -class MaxRetriesExceededError(TaskitoError): +class MaxRetriesExceededError(TaskitoError, JobErrorDetails): """Raised when a task has exhausted all retry attempts.""" diff --git a/sdks/python/taskito/prefork/child.py b/sdks/python/taskito/prefork/child.py index d4331741..8fd28dd3 100644 --- a/sdks/python/taskito/prefork/child.py +++ b/sdks/python/taskito/prefork/child.py @@ -38,6 +38,7 @@ ) from taskito.exceptions import TaskCancelledError from taskito.log_config import silence_asyncio_pipe_noise +from taskito.task_errors import encode_task_error logger = logging.getLogger("taskito.prefork.child") @@ -141,13 +142,14 @@ def _execute_job( except Exception: wall_time_ns = time.monotonic_ns() - start_ns - error_msg = traceback.format_exc() - logger.error("task %s[%s] failed: %s", task_name, job_id, error_msg.splitlines()[-1]) + exc = sys.exc_info()[1] + error_msg = encode_task_error(exc) if exc is not None else traceback.format_exc() + last_line = traceback.format_exc().splitlines()[-1] + logger.error("task %s[%s] failed: %s", task_name, job_id, last_line) should_retry = True filters = queue._task_retry_filters.get(task_name) if filters: - exc = sys.exc_info()[1] dont_retry_on = filters.get("dont_retry_on", []) for cls in dont_retry_on: if isinstance(exc, cls): diff --git a/sdks/python/taskito/result.py b/sdks/python/taskito/result.py index 82b5bde5..ce062963 100644 --- a/sdks/python/taskito/result.py +++ b/sdks/python/taskito/result.py @@ -14,7 +14,9 @@ SerializationError, TaskCancelledError, TaskFailedError, + TaskitoError, ) +from taskito.task_errors import decode_task_error if TYPE_CHECKING: from taskito._taskito import PyJob @@ -29,19 +31,24 @@ def _summarize_error(error: str | None) -> str | None: - """Reduce a stored traceback to its final line (exception type + message). + """Reduce a stored error to one ``ExceptionType: message`` line. - A Python traceback ends with the ``ExceptionType: message`` line; the - intervening frames carry file paths and source snippets we don't want in - the broadly-readable job list. Return that last non-empty line, capped. + Structured errors (canonical JSON, see ``taskito.task_errors``) summarize + from their fields; legacy plain tracebacks fall back to the last non-empty + line. Frames carry file paths and source snippets we don't want in the + broadly-readable job list, so only the summary is surfaced here. """ if not error: return error - last_line = next((ln for ln in reversed(error.splitlines()) if ln.strip()), error) - last_line = last_line.strip() - if len(last_line) > _ERROR_SUMMARY_MAX: - last_line = last_line[:_ERROR_SUMMARY_MAX] + "…" - return last_line + decoded = decode_task_error(error) + if decoded is not None: + summary = decoded.summary() + else: + summary = next((ln for ln in reversed(error.splitlines()) if ln.strip()), error) + summary = summary.strip() + if len(summary) > _ERROR_SUMMARY_MAX: + summary = summary[:_ERROR_SUMMARY_MAX] + "…" + return summary class JobResult(AsyncJobResultMixin): @@ -156,14 +163,33 @@ def _poll_once(self) -> tuple[str, Any]: error_msg = self._py_job.error or "job was cancelled" raise TaskCancelledError(f"Job {self.id} cancelled: {error_msg}") if status == "dead": - error_msg = self._py_job.error or "max retries exceeded" - raise MaxRetriesExceededError(f"Job {self.id} dead-lettered: {error_msg}") + raise self._failure_error( + MaxRetriesExceededError, "dead-lettered", "max retries exceeded" + ) if status == "failed": - error_msg = self._py_job.error or "task failed" - raise TaskFailedError(f"Job {self.id} failed: {error_msg}") + raise self._failure_error(TaskFailedError, "failed", "task failed") return status, None + def _failure_error( + self, + error_cls: type[TaskFailedError] | type[MaxRetriesExceededError], + verb: str, + default_msg: str, + ) -> TaskitoError: + """Build the outcome exception, attaching structured details when stored.""" + raw = self._py_job.error + decoded = decode_task_error(raw) + summary = decoded.summary() if decoded is not None else (raw or default_msg) + error = error_cls(f"Job {self.id} {verb}: {summary}") + error._attach_details( + errtype=decoded.errtype if decoded else None, + traceback=decoded.traceback if decoded else None, + job_id=self.id, + raw_error=raw, + ) + return error + def result( self, timeout: float = 30.0, diff --git a/sdks/python/taskito/task_errors.py b/sdks/python/taskito/task_errors.py new file mode 100644 index 00000000..b02a4640 --- /dev/null +++ b/sdks/python/taskito/task_errors.py @@ -0,0 +1,70 @@ +"""Canonical structured task-error encoding (cross-SDK contract). + +A task failure is stored as one JSON object — ``{"errtype", "message", +"traceback"}`` — so any Taskito SDK (and the dashboard) can show the exception +class and message without parsing language-specific traceback text. Strings +that don't parse as that object are legacy/system errors and pass through +verbatim; see ``BINDING_CONTRACT.md`` "Task errors". +""" + +from __future__ import annotations + +import json +import traceback as traceback_module +from dataclasses import dataclass +from types import TracebackType + + +@dataclass(frozen=True) +class TaskError: + """A decoded structured task error.""" + + errtype: str + message: str + traceback: list[str] + raw: str + + def summary(self) -> str: + """One-line ``errtype: message`` form used anywhere errors are listed.""" + return f"{self.errtype}: {self.message}" if self.message else self.errtype + + +def encode_task_error(exc: BaseException) -> str: + """Encode a raised exception as the canonical cross-SDK JSON string.""" + return encode_from_parts(type(exc), exc, exc.__traceback__) + + +def encode_from_parts( + exc_type: type[BaseException], + exc: BaseException, + tb: TracebackType | None, +) -> str: + """Encode from explicit ``(type, value, traceback)`` parts. + + Separate entry point because the Rust worker paths hold the three parts + individually (PyO3 ``PyErr``) rather than a single raised exception. + """ + frames = "".join(traceback_module.format_exception(exc_type, exc, tb)).splitlines() + return json.dumps( + {"errtype": exc_type.__qualname__, "message": str(exc), "traceback": frames}, + separators=(",", ":"), + ) + + +def decode_task_error(raw: str | None) -> TaskError | None: + """Decode a stored error string, or None for legacy/system plain strings.""" + if not raw or not raw.startswith("{"): + return None + try: + decoded = json.loads(raw) + except ValueError: + return None + if not isinstance(decoded, dict) or not isinstance(decoded.get("message"), str): + return None + tb = decoded.get("traceback") + return TaskError( + errtype=str(decoded.get("errtype") or "Error"), + message=decoded["message"], + traceback=[str(line) for line in tb] if isinstance(tb, list) else [], + raw=raw, + ) diff --git a/sdks/python/tests/core/test_task_errors.py b/sdks/python/tests/core/test_task_errors.py new file mode 100644 index 00000000..c55c931e --- /dev/null +++ b/sdks/python/tests/core/test_task_errors.py @@ -0,0 +1,128 @@ +"""Structured task-error encoding/decoding and end-to-end propagation.""" + +import json +import threading +from typing import Any + +import pytest + +from taskito import Queue +from taskito.exceptions import MaxRetriesExceededError, TaskFailedError +from taskito.result import _summarize_error +from taskito.task_errors import decode_task_error, encode_from_parts, encode_task_error + + +def _raise_and_capture() -> BaseException: + try: + raise ValueError("bad value 42") + except ValueError as exc: + return exc + + +class TestEncode: + def test_matches_contract_vector(self) -> None: + """Byte-exact against the BINDING_CONTRACT.md test vector.""" + + class BoomError(Exception): + pass + + exc = BoomError("it broke") + encoded = encode_from_parts(BoomError, exc, None) + decoded = json.loads(encoded) + decoded["traceback"] = ["frame1", "frame2"] + assert ( + json.dumps(decoded, separators=(",", ":")) + == '{"errtype":"TestEncode.test_matches_contract_vector..BoomError",' + '"message":"it broke","traceback":["frame1","frame2"]}' + ) + + def test_encodes_type_message_and_frames(self) -> None: + exc = _raise_and_capture() + decoded = json.loads(encode_task_error(exc)) + assert decoded["errtype"] == "ValueError" + assert decoded["message"] == "bad value 42" + assert any("raise ValueError" in line for line in decoded["traceback"]) + + def test_key_order_and_compactness(self) -> None: + exc = _raise_and_capture() + encoded = encode_task_error(exc) + assert encoded.startswith('{"errtype":"ValueError","message":"bad value 42","traceback":[') + assert ": " not in encoded.split('"traceback"')[0] + + +class TestDecode: + def test_round_trip(self) -> None: + exc = _raise_and_capture() + decoded = decode_task_error(encode_task_error(exc)) + assert decoded is not None + assert decoded.errtype == "ValueError" + assert decoded.message == "bad value 42" + assert decoded.summary() == "ValueError: bad value 42" + + @pytest.mark.parametrize( + "raw", + [ + None, + "", + "plain legacy traceback\nValueError: nope", + "job timed out after 5000ms", + "[1, 2]", + '{"no_message_key": true}', + '{"message": 42}', + "{not json", + ], + ) + def test_legacy_and_malformed_return_none(self, raw: str | None) -> None: + assert decode_task_error(raw) is None + + +class TestSummarize: + def test_structured_error_summarized_from_fields(self) -> None: + exc = _raise_and_capture() + assert _summarize_error(encode_task_error(exc)) == "ValueError: bad value 42" + + def test_legacy_traceback_still_uses_last_line(self) -> None: + tb = "Traceback (most recent call last):\n ...\nValueError: legacy" + assert _summarize_error(tb) == "ValueError: legacy" + + +class TestEndToEnd: + def test_failed_job_stores_structured_error( + self, queue: Queue, run_worker: threading.Thread, poll_until: Any + ) -> None: + @queue.task(max_retries=0) + def explode() -> None: + raise RuntimeError("kaboom") + + result = explode.delay() + + def job_failed() -> bool: + job = queue.get_job(result.id) + return job is not None and job.status in ("failed", "dead") + + poll_until(job_failed, message="job should fail") + job = queue.get_job(result.id) + assert job is not None + decoded = decode_task_error(job.error) + assert decoded is not None + assert decoded.errtype == "RuntimeError" + assert decoded.message == "kaboom" + assert decoded.traceback + + def test_result_raises_with_structured_details( + self, queue: Queue, run_worker: threading.Thread + ) -> None: + @queue.task(max_retries=0) + def explode() -> None: + raise RuntimeError("kaboom") + + result = explode.delay() + with pytest.raises((TaskFailedError, MaxRetriesExceededError)) as excinfo: + result.result(timeout=10) + error = excinfo.value + assert isinstance(error, (TaskFailedError, MaxRetriesExceededError)) + assert "RuntimeError: kaboom" in str(error) + assert error.errtype == "RuntimeError" + assert error.job_id == result.id + assert error.traceback + assert error.raw_error is not None and error.raw_error.startswith("{") From 2254fd41ff9fa0dd0616a6456fa55200c30828b9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:21:56 +0530 Subject: [PATCH 3/8] feat(node): emit and decode structured task errors --- crates/taskito-node/src/dispatcher.rs | 10 ++- sdks/node/src/errors.ts | 27 +++++++- sdks/node/src/index.ts | 1 + sdks/node/src/task-error.ts | 60 ++++++++++++++++++ sdks/node/src/worker.ts | 8 ++- sdks/node/test/core/task-error.test.ts | 86 ++++++++++++++++++++++++++ 6 files changed, 188 insertions(+), 4 deletions(-) create mode 100644 sdks/node/src/task-error.ts create mode 100644 sdks/node/test/core/task-error.test.ts diff --git a/crates/taskito-node/src/dispatcher.rs b/crates/taskito-node/src/dispatcher.rs index 6223ff42..42f8b2c1 100644 --- a/crates/taskito-node/src/dispatcher.rs +++ b/crates/taskito-node/src/dispatcher.rs @@ -119,7 +119,15 @@ async fn run_one(callback: &TaskCallback, storage: &StorageBackend, mut job: Job wall_time_ns, } } else { - failure(job, err.to_string(), wall_time_ns, false) + // `Error::to_string()` prepends the napi status ("GenericFailure, "); + // the bare reason is the JS error's string form, which the worker + // shapes into the cross-SDK structured-error JSON. + let reason = if err.reason.is_empty() { + err.to_string() + } else { + err.reason.clone() + }; + failure(job, reason, wall_time_ns, false) } } Ok(Err(_)) => failure( diff --git a/sdks/node/src/errors.ts b/sdks/node/src/errors.ts index cb93697f..db229355 100644 --- a/sdks/node/src/errors.ts +++ b/sdks/node/src/errors.ts @@ -1,3 +1,5 @@ +import { decodeTaskError } from "./task-error"; + /** Base class for all Taskito SDK errors. Every error below extends this. */ export class TaskitoError extends Error { constructor(message: string) { @@ -16,14 +18,35 @@ export class TaskNotRegisteredError extends TaskitoError { } } -/** Thrown by {@link Queue.result} when the awaited job failed or dead-lettered. */ +/** + * Thrown by {@link Queue.result} when the awaited job failed or dead-lettered. + * A structured (cross-SDK JSON) reason exposes `errtype`/`traceback`; a plain + * legacy/system string is surfaced verbatim with those fields undefined. + */ export class JobFailedError extends TaskitoError { + /** The task exception's class name, when the reason is structured. */ + readonly errtype?: string; + /** Best-effort stack lines from the failing worker, when structured. */ + readonly traceback?: readonly string[]; + /** The stored error string exactly as persisted. */ + readonly raw: string; + constructor( readonly jobId: string, reason: string, ) { - super(`Job ${jobId} failed: ${reason}`); + const decoded = decodeTaskError(reason); + super( + decoded + ? `Job ${jobId} failed: ${decoded.errtype}: ${decoded.message}` + : `Job ${jobId} failed: ${reason}`, + ); this.name = "JobFailedError"; + this.raw = reason; + if (decoded) { + this.errtype = decoded.errtype; + this.traceback = decoded.traceback; + } } } diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 05cb438f..6cfc2096 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -75,6 +75,7 @@ export { type Serializer, SignedSerializer, } from "./serializers"; +export { decodeTaskError, encodeTaskError, type TaskError } from "./task-error"; export type { AnyHandler, CircuitBreakerOptions, diff --git a/sdks/node/src/task-error.ts b/sdks/node/src/task-error.ts new file mode 100644 index 00000000..63de1d96 --- /dev/null +++ b/sdks/node/src/task-error.ts @@ -0,0 +1,60 @@ +/** + * Canonical structured task-error format (cross-SDK contract). + * + * A failed task is stored as one JSON object — `{"errtype", "message", + * "traceback"}` in that key order — so any SDK can read another's failures. + * Plain strings (core maintenance errors, legacy rows) stay untouched; + * readers fall back via {@link decodeTaskError} returning null. + */ + +/** A decoded structured task error. */ +export interface TaskError { + errtype: string; + message: string; + traceback: string[]; +} + +/** Encode a thrown value as the canonical cross-SDK error JSON. */ +export function encodeTaskError(error: unknown): string { + let errtype = "Error"; + let message: string; + let traceback: string[] = []; + if (error instanceof Error) { + errtype = error.name || error.constructor.name || "Error"; + message = error.message ?? String(error); + // Keep stack lines verbatim — readers render them as-is. + traceback = error.stack ? error.stack.split("\n") : []; + } else { + message = String(error); + } + // Object-literal key order is the contract's key order. + return JSON.stringify({ errtype, message, traceback }); +} + +/** + * Decode a stored error string. Returns null when it is not a structured + * error (the contract's fallback rule: not a JSON object with a string + * `message`), in which case callers surface the raw string unchanged. + */ +export function decodeTaskError(raw: string): TaskError | null { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return null; + } + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + return null; + } + const candidate = parsed as Record; + if (typeof candidate.message !== "string") { + return null; + } + return { + errtype: typeof candidate.errtype === "string" ? candidate.errtype : "Error", + message: candidate.message, + traceback: Array.isArray(candidate.traceback) + ? candidate.traceback.filter((line): line is string => typeof line === "string") + : [], + }; +} diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 73ef2568..7b78c165 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -20,6 +20,7 @@ import type { } from "./native"; import { type ResourceRuntime, runWithResolver } from "./resources"; import { deserializeCall, type PayloadCodec, type Serializer } from "./serializers"; +import { encodeTaskError } from "./task-error"; import type { QueueLimits, RegisteredTask, WorkerRunOptions } from "./types"; import { createLogger } from "./utils"; import type { WorkflowTracker } from "./workflows"; @@ -175,7 +176,12 @@ export class Worker { // onError hooks must not mask the original task failure. } } - throw error; + // Rethrow as the canonical structured-error JSON. With an empty name, + // Error#toString() is just the message, so the native layer stores the + // bare JSON object as the job's error string. + const structured = new Error(encodeTaskError(error)); + structured.name = ""; + throw structured; } finally { clearInterval(poller); try { diff --git a/sdks/node/test/core/task-error.test.ts b/sdks/node/test/core/task-error.test.ts new file mode 100644 index 00000000..59baa178 --- /dev/null +++ b/sdks/node/test/core/task-error.test.ts @@ -0,0 +1,86 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { decodeTaskError, encodeTaskError, JobFailedError, Queue, type Worker } from "../../src"; + +let worker: Worker | undefined; +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function newQueue(): Queue { + return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-terr-")), "q.db") }); +} + +class BoomError extends Error { + constructor(message: string) { + super(message); + this.name = "BoomError"; + } +} + +it("encodeTaskError matches the contract test vector byte-exact", () => { + const error = new BoomError("it broke"); + error.stack = "frame1\nframe2"; + expect(encodeTaskError(error)).toBe( + '{"errtype":"BoomError","message":"it broke","traceback":["frame1","frame2"]}', + ); +}); + +it("encodeTaskError handles non-Error thrown values", () => { + expect(decodeTaskError(encodeTaskError("oops"))).toEqual({ + errtype: "Error", + message: "oops", + traceback: [], + }); +}); + +it("decodeTaskError returns null for anything but an object with a string message", () => { + expect(decodeTaskError("plain failure string")).toBeNull(); + expect(decodeTaskError('["not","an","object"]')).toBeNull(); + expect(decodeTaskError('{"errtype":"X","traceback":[]}')).toBeNull(); +}); + +it("stores a failing task's error as canonical JSON and surfaces it structured", async () => { + const queue = newQueue(); + queue.task( + "boom", + () => { + throw new BoomError("it broke"); + }, + { maxRetries: 0 }, + ); + const id = queue.enqueue("boom"); + worker = queue.runWorker(); + + let caught: unknown; + try { + await queue.result(id); + } catch (error) { + caught = error; + } + expect(caught).toBeInstanceOf(JobFailedError); + const failed = caught as JobFailedError; + + // The stored string must be the bare JSON object — nothing prefixed by the + // native layer. + const decoded = decodeTaskError(failed.raw); + expect(decoded).not.toBeNull(); + expect(decoded?.errtype).toBe("BoomError"); + expect(decoded?.message).toBe("it broke"); + expect(decoded?.traceback.length).toBeGreaterThan(0); + + expect(failed.errtype).toBe("BoomError"); + expect(failed.traceback?.length).toBeGreaterThan(0); + expect(failed.message).toBe(`Job ${id} failed: BoomError: it broke`); +}); + +it("JobFailedError falls back verbatim on a plain-string reason", () => { + const failed = new JobFailedError("j1", "task timed out"); + expect(failed.errtype).toBeUndefined(); + expect(failed.traceback).toBeUndefined(); + expect(failed.raw).toBe("task timed out"); + expect(failed.message).toBe("Job j1 failed: task timed out"); +}); From a8be0a94f185cabdfa4461d6956c95bbb98124ba Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:22:02 +0530 Subject: [PATCH 4/8] feat(java): emit and decode structured task errors --- .../taskito/contrib/SentryMiddleware.java | 5 +- .../byteveda/taskito/errors/TaskError.java | 31 +++++++ .../byteveda/taskito/errors/TaskErrors.java | 86 +++++++++++++++++++ .../byteveda/taskito/errors/package-info.java | 4 + .../taskito/worker/WorkerDispatchBridge.java | 9 +- .../taskito/errors/TaskErrorsTest.java | 82 ++++++++++++++++++ .../test/InMemoryQueueBackendTest.java | 12 ++- 7 files changed, 221 insertions(+), 8 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/contrib/SentryMiddleware.java b/sdks/java/src/main/java/org/byteveda/taskito/contrib/SentryMiddleware.java index 0447223b..77f56bd3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/contrib/SentryMiddleware.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/contrib/SentryMiddleware.java @@ -3,6 +3,7 @@ import io.sentry.Sentry; import io.sentry.SentryLevel; import java.util.function.Predicate; +import org.byteveda.taskito.errors.TaskErrors; import org.byteveda.taskito.events.OutcomeEvent; import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.middleware.TaskContext; @@ -46,7 +47,9 @@ public void onDeadLetter(OutcomeEvent event) { scope.setLevel(SentryLevel.FATAL); scope.setTag("taskito.task", event.taskName); scope.setTag("taskito.job", event.jobId); - Sentry.captureMessage("task dead-lettered: " + (event.error == null ? "" : event.error)); + // Stored errors may be canonical JSON; headline with "errtype: message" when so. + Sentry.captureMessage( + "task dead-lettered: " + (event.error == null ? "" : TaskErrors.summarize(event.error))); }); } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java new file mode 100644 index 00000000..c14c1f01 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java @@ -0,0 +1,31 @@ +package org.byteveda.taskito.errors; + +import java.util.List; + +/** + * A structured task error decoded from the canonical JSON stored in {@code error} + * fields (jobs, error history, dead letter). Obtain via + * {@link TaskErrors#decode(String)}; plain system strings decode to {@code null}. + */ +public final class TaskError { + /** Exception class name reported by the producing worker; empty when absent. */ + public final String errtype; + /** Human-readable message, verbatim; may be empty. */ + public final String message; + /** Best-effort stack frames, throw site first; empty when unavailable. */ + public final List traceback; + /** The stored string this was decoded from. */ + public final String raw; + + TaskError(String errtype, String message, List traceback, String raw) { + this.errtype = errtype; + this.message = message; + this.traceback = List.copyOf(traceback); + this.raw = raw; + } + + /** One-line human summary: {@code errtype: message}, or just the message when errtype is empty. */ + public String summary() { + return errtype.isEmpty() ? message : errtype + ": " + message; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java new file mode 100644 index 00000000..f13baf26 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java @@ -0,0 +1,86 @@ +package org.byteveda.taskito.errors; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * Codec for the canonical cross-SDK task-error JSON + * ({@code {"errtype":...,"message":...,"traceback":[...]}}, keys in that order, + * no extra whitespace) stored verbatim in job, error-history, and dead-letter + * {@code error} fields. Core-generated maintenance errors (timeout, expiry, + * worker-death recovery) stay plain strings by design, so readers must fall + * back — {@link #decode(String)} returns {@code null} for those. + */ +public final class TaskErrors { + private static final ObjectMapper JSON = new ObjectMapper(); + + private TaskErrors() {} + + /** Encode a thrown task error: fully-qualified class name, verbatim message, stack frames in order. */ + public static String encode(Throwable error) { + ObjectNode node = JSON.createObjectNode(); + node.put("errtype", error.getClass().getName()); + String message = error.getMessage(); + node.put("message", message == null ? "" : message); + ArrayNode traceback = node.putArray("traceback"); + for (StackTraceElement frame : error.getStackTrace()) { + traceback.add(frame.toString()); + } + return node.toString(); + } + + /** + * Decode a stored error string, e.g. {@link org.byteveda.taskito.events.OutcomeEvent#error} + * or {@link org.byteveda.taskito.model.DeadJob#error}. Returns {@code null} unless the + * string is a JSON object with a string {@code message} — the signal that it is plain + * legacy/system text and must be surfaced as-is. + */ + public static TaskError decode(String raw) { + if (raw == null || raw.isEmpty()) { + return null; + } + JsonNode node = readTree(raw); + if (node == null || !node.isObject()) { + return null; + } + JsonNode message = node.get("message"); + if (message == null || !message.isTextual()) { + return null; + } + JsonNode errtype = node.get("errtype"); + String errtypeText = errtype != null && errtype.isTextual() ? errtype.asText() : ""; + return new TaskError(errtypeText, message.asText(), readTraceback(node.get("traceback")), raw); + } + + /** One-line human summary: {@code errtype: message} when structured, the raw string otherwise. */ + public static String summarize(String raw) { + TaskError decoded = decode(raw); + return decoded == null ? raw : decoded.summary(); + } + + private static JsonNode readTree(String raw) { + try { + return JSON.readTree(raw); + } catch (IOException e) { + return null; + } + } + + private static List readTraceback(JsonNode node) { + if (node == null || !node.isArray()) { + return List.of(); + } + List frames = new ArrayList<>(node.size()); + for (JsonNode frame : node) { + if (frame.isTextual()) { + frames.add(frame.asText()); + } + } + return frames; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java index 79b9b402..24eccaa6 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java @@ -6,5 +6,9 @@ * ({@link org.byteveda.taskito.errors.SerializationException}, * {@link org.byteveda.taskito.errors.WorkflowException}, etc.) to react to one * category. Native (JNI) errors surface as the base {@code TaskitoException}. + * + *

Also home to {@link org.byteveda.taskito.errors.TaskErrors}, the codec for + * the structured task-error JSON stored in job and dead-letter {@code error} + * fields, and its decoded view {@link org.byteveda.taskito.errors.TaskError}. */ package org.byteveda.taskito.errors; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index 5bc6dc35..b64528de 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -8,6 +8,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import org.byteveda.taskito.errors.TaskErrors; import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; @@ -103,7 +104,8 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { for (Middleware m : middleware) { m.onError(context, t); } - bound.failJob(token, describe(t)); + // Canonical structured error (middleware above saw the live Throwable). + bound.failJob(token, TaskErrors.encode(t)); } finally { if (scope != null) { Resources.exit(scope); // unbind the thread + dispose task-scoped resources (LIFO) @@ -183,9 +185,4 @@ private static JsonNode readTree(String json) { return null; } } - - private static String describe(Throwable t) { - String message = t.getMessage(); - return message == null ? t.getClass().getSimpleName() : t.getClass().getSimpleName() + ": " + message; - } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java b/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java new file mode 100644 index 00000000..65690ece --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java @@ -0,0 +1,82 @@ +package org.byteveda.taskito.errors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.util.List; +import org.junit.jupiter.api.Test; + +class TaskErrorsTest { + + /** Named so the fully-qualified {@code errtype} assertion below stays stable. */ + static final class BoomError extends RuntimeException { + BoomError(String message) { + super(message); + } + } + + private static BoomError boom(String message) { + BoomError error = new BoomError(message); + error.setStackTrace(new StackTraceElement[] { + new StackTraceElement("Frame", "one", "Frame.java", 1), + new StackTraceElement("Frame", "two", "Frame.java", 2), + }); + return error; + } + + @Test + void encodeMatchesContractVectorShape() { + // BINDING_CONTRACT.md vector semantics: errtype/message/traceback keys in + // that order, compact JSON, frames in getStackTrace() order. + String expected = "{\"errtype\":\"" + BoomError.class.getName() + + "\",\"message\":\"it broke\"," + + "\"traceback\":[\"Frame.one(Frame.java:1)\",\"Frame.two(Frame.java:2)\"]}"; + assertEquals(expected, TaskErrors.encode(boom("it broke"))); + } + + @Test + void encodeNullMessageBecomesEmptyString() { + String expected = "{\"errtype\":\"" + BoomError.class.getName() + + "\",\"message\":\"\"," + + "\"traceback\":[\"Frame.one(Frame.java:1)\",\"Frame.two(Frame.java:2)\"]}"; + assertEquals(expected, TaskErrors.encode(boom(null))); + } + + @Test + void decodeContractVectorLiteral() { + String raw = "{\"errtype\":\"BoomError\",\"message\":\"it broke\",\"traceback\":[\"frame1\",\"frame2\"]}"; + TaskError decoded = TaskErrors.decode(raw); + assertEquals("BoomError", decoded.errtype); + assertEquals("it broke", decoded.message); + assertEquals(List.of("frame1", "frame2"), decoded.traceback); + assertEquals(raw, decoded.raw); + } + + @Test + void decodeRoundTripsEncode() { + TaskError decoded = TaskErrors.decode(TaskErrors.encode(boom("it broke"))); + assertEquals(BoomError.class.getName(), decoded.errtype); + assertEquals("it broke", decoded.message); + assertEquals(List.of("Frame.one(Frame.java:1)", "Frame.two(Frame.java:2)"), decoded.traceback); + } + + @Test + void decodeFallsBackToNullForUnstructuredStrings() { + assertNull(TaskErrors.decode("worker died mid-flight")); // plain system string + assertNull(TaskErrors.decode("[\"frame1\"]")); // JSON but not an object + assertNull(TaskErrors.decode("{\"errtype\":\"BoomError\"}")); // object without message + assertNull(TaskErrors.decode("{\"message\":42}")); // message not a string + assertNull(TaskErrors.decode("")); + assertNull(TaskErrors.decode(null)); + } + + @Test + void summarizePrefersErrtypeMessageHeadline() { + assertEquals( + "BoomError: it broke", + TaskErrors.summarize("{\"errtype\":\"BoomError\",\"message\":\"it broke\",\"traceback\":[]}")); + assertEquals("just text", TaskErrors.summarize("{\"message\":\"just text\"}")); // no errtype + assertEquals( + "task timed out after 5000ms", TaskErrors.summarize("task timed out after 5000ms")); // raw fallback + } +} diff --git a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java index a7b91538..1f1c2bff 100644 --- a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java +++ b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryQueueBackendTest.java @@ -1,6 +1,7 @@ package org.byteveda.taskito.test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import java.time.Duration; @@ -8,6 +9,9 @@ import java.util.Collections; import java.util.List; import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.errors.TaskError; +import org.byteveda.taskito.errors.TaskErrors; +import org.byteveda.taskito.model.DeadJob; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobStatus; import org.byteveda.taskito.task.Task; @@ -71,7 +75,13 @@ void retriesThenDeadLetters() throws Exception { Job job = queue.awaitJob(id, Duration.ofSeconds(10)).orElseThrow(); assertEquals(JobStatus.DEAD, job.status); assertEquals(2, job.retryCount); - assertTrue(queue.listDead(10, 0).size() >= 1); + List dead = queue.listDead(10, 0); + assertTrue(dead.size() >= 1); + // The bridge stores the canonical structured error; parity with the JNI path. + TaskError error = TaskErrors.decode(dead.get(0).error); + assertEquals(IllegalStateException.class.getName(), error.errtype); + assertEquals("boom", error.message); + assertFalse(error.traceback.isEmpty()); } } } From 3abfd38ba2f6da72f198ccd6933fdbf50d4307e2 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:22:02 +0530 Subject: [PATCH 5/8] feat(dashboard): render structured task errors --- dashboard/src/components/ui/index.ts | 1 + dashboard/src/components/ui/task-error.tsx | 50 ++++++++ .../components/dead-letter-row.tsx | 6 +- .../components/dead-letter-table.tsx | 8 +- .../src/features/dead-letters/utils.test.ts | 14 +++ dashboard/src/features/dead-letters/utils.ts | 17 ++- .../jobs/components/job-errors-tab.tsx | 6 +- .../jobs/components/job-overview-tab.tsx | 13 +- .../features/jobs/components/job-table.tsx | 8 +- .../components/workflow-nodes-table.tsx | 4 +- dashboard/src/lib/task-error.test.ts | 113 ++++++++++++++++++ dashboard/src/lib/task-error.ts | 62 ++++++++++ dashboard/src/routes/workflows_.$id.tsx | 6 +- 13 files changed, 282 insertions(+), 26 deletions(-) create mode 100644 dashboard/src/components/ui/task-error.tsx create mode 100644 dashboard/src/lib/task-error.test.ts create mode 100644 dashboard/src/lib/task-error.ts diff --git a/dashboard/src/components/ui/index.ts b/dashboard/src/components/ui/index.ts index 0ab48c0c..97329795 100644 --- a/dashboard/src/components/ui/index.ts +++ b/dashboard/src/components/ui/index.ts @@ -103,5 +103,6 @@ export { TableRow, } from "./table"; export { Tabs, TabsContent, TabsList, TabsTrigger } from "./tabs"; +export { TaskErrorBlock, TaskErrorSummary } from "./task-error"; export { Toaster, toast } from "./toaster"; export { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "./tooltip"; diff --git a/dashboard/src/components/ui/task-error.tsx b/dashboard/src/components/ui/task-error.tsx new file mode 100644 index 00000000..92a59906 --- /dev/null +++ b/dashboard/src/components/ui/task-error.tsx @@ -0,0 +1,50 @@ +import { cn } from "@/lib/cn"; +import { parseTaskError, taskErrorSummary, taskErrorTooltip } from "@/lib/task-error"; + +/** + * Full task-error block for detail views. Structured errors get an + * `errtype: message` headline plus the traceback; legacy strings render + * verbatim. `className` carries per-site sizing/background (max-h, bg, + * padding) so each call site keeps its existing look. + */ +export function TaskErrorBlock({ error, className }: { error: string; className?: string }) { + const parsed = parseTaskError(error); + + if (!parsed) { + return ( + 

+        {error}
+
+ ); + } + + return ( +
+
{taskErrorSummary(parsed)}
+ {parsed.traceback.length > 0 ? ( + 
{parsed.traceback.join("\n")}
+ ) : null} +
+ ); +} + +/** + * One-line task-error for table cells: structured → `errtype: message` with + * the traceback tail in the title tooltip; legacy → the raw string with the + * full text as title (the pre-existing behavior). + */ +export function TaskErrorSummary({ error, className }: { error: string; className?: string }) { + const parsed = parseTaskError(error); + const text = parsed ? taskErrorSummary(parsed) : error; + const title = parsed ? taskErrorTooltip(parsed) : error; + return ( + + {text} + + ); +} diff --git a/dashboard/src/features/dead-letters/components/dead-letter-row.tsx b/dashboard/src/features/dead-letters/components/dead-letter-row.tsx index f8b3cd9d..087fc254 100644 --- a/dashboard/src/features/dead-letters/components/dead-letter-row.tsx +++ b/dashboard/src/features/dead-letters/components/dead-letter-row.tsx @@ -1,6 +1,6 @@ import { Link } from "@tanstack/react-router"; import { RotateCcw, Trash2 } from "lucide-react"; -import { Badge, Button } from "@/components/ui"; +import { Badge, Button, TaskErrorBlock } from "@/components/ui"; import type { DeadLetter } from "@/lib/api-types"; import { formatRelative } from "@/lib/time"; import { useDeleteDeadLetter, useRetryDeadLetter } from "../hooks"; @@ -39,9 +39,7 @@ export function DeadLetterRow({ item }: DeadLetterRowProps) { {item.error ? ( - 
-            {item.error}
-
+ ) : null}
diff --git a/dashboard/src/features/dead-letters/components/dead-letter-table.tsx b/dashboard/src/features/dead-letters/components/dead-letter-table.tsx index 2428cd36..a8c0cd2e 100644 --- a/dashboard/src/features/dead-letters/components/dead-letter-table.tsx +++ b/dashboard/src/features/dead-letters/components/dead-letter-table.tsx @@ -9,6 +9,7 @@ import { TableHead, TableHeader, TableRow, + TaskErrorSummary, } from "@/components/ui"; import type { DeadLetter } from "@/lib/api-types"; import { formatRelative } from "@/lib/time"; @@ -60,9 +61,10 @@ function DeadLetterTableRow({ item }: { item: DeadLetter }) { {item.queue} {item.error ? ( - - {item.error} - + ) : ( — )} diff --git a/dashboard/src/features/dead-letters/utils.test.ts b/dashboard/src/features/dead-letters/utils.test.ts index dd626285..a0df954e 100644 --- a/dashboard/src/features/dead-letters/utils.test.ts +++ b/dashboard/src/features/dead-letters/utils.test.ts @@ -30,12 +30,19 @@ const TRACEBACK_CONNECTION_ERROR = `Traceback (most recent call last): raise ConnectionError(f"upstream {endpoint} unreachable") ConnectionError: upstream /v1/users unreachable`; +const STRUCTURED_ERROR = + '{"errtype":"BoomError","message":"it broke","traceback":["frame1","frame2"]}'; + describe("extractExceptionClass", () => { it("pulls the class name from a standard traceback", () => { expect(extractExceptionClass(TRACEBACK_VALUE_ERROR)).toBe("ValueError"); expect(extractExceptionClass(TRACEBACK_CONNECTION_ERROR)).toBe("ConnectionError"); }); + it("uses errtype for structured errors", () => { + expect(extractExceptionClass(STRUCTURED_ERROR)).toBe("BoomError"); + }); + it("handles custom exception class names", () => { const traceback = "something\nMyCustomError: oops"; expect(extractExceptionClass(traceback)).toBe("MyCustomError"); @@ -61,6 +68,13 @@ describe("extractReason", () => { expect(extractReason(null)).toMatch(/no error captured/i); expect(extractReason(" ")).toMatch(/no error captured/i); }); + + it("uses message for structured errors", () => { + expect(extractReason(STRUCTURED_ERROR)).toBe("it broke"); + expect(extractReason('{"errtype":"E","message":"","traceback":[]}')).toMatch( + /no error captured/i, + ); + }); }); describe("groupByError", () => { diff --git a/dashboard/src/features/dead-letters/utils.ts b/dashboard/src/features/dead-letters/utils.ts index ab7294eb..5b418c16 100644 --- a/dashboard/src/features/dead-letters/utils.ts +++ b/dashboard/src/features/dead-letters/utils.ts @@ -1,4 +1,5 @@ import type { DeadLetter } from "@/lib/api-types"; +import { parseTaskError } from "@/lib/task-error"; export interface DeadLetterGroup { /** Stable key: `taskName::ExceptionClass`. */ @@ -21,14 +22,17 @@ const EMPTY_ERROR_LABEL = "(no error captured)"; const UNKNOWN_EXCEPTION = "Error"; /** - * Extract the exception class from a Python traceback string. + * Extract the exception class from an error string. * - * Python tracebacks end with a line like ``ValueError: permanent failure…``. - * We read the last non-blank line and pull the token before the first `:`. - * Falls back to "Error" if the input isn't a recognisable traceback. + * Structured errors (canonical JSON per the cross-SDK contract) carry the + * class in `errtype`. Legacy Python tracebacks end with a line like + * ``ValueError: permanent failure…`` — we read the last non-blank line and + * pull the token before the first `:`. Falls back to "Error" otherwise. */ export function extractExceptionClass(error: string | null | undefined): string { if (!error) return UNKNOWN_EXCEPTION; + const structured = parseTaskError(error); + if (structured) return structured.errtype; const trimmed = error.trim(); if (!trimmed) return UNKNOWN_EXCEPTION; const lines = trimmed.split("\n"); @@ -47,10 +51,13 @@ export function extractExceptionClass(error: string | null | undefined): string /** * Extract the one-line reason from a traceback — the text after the exception - * class on the final traceback line. Useful for a group's sub-summary. + * class on the final traceback line (or `message` for structured errors). + * Useful for a group's sub-summary. */ export function extractReason(error: string | null | undefined): string { if (!error) return EMPTY_ERROR_LABEL; + const structured = parseTaskError(error); + if (structured) return structured.message || EMPTY_ERROR_LABEL; const trimmed = error.trim(); if (!trimmed) return EMPTY_ERROR_LABEL; const lines = trimmed.split("\n"); diff --git a/dashboard/src/features/jobs/components/job-errors-tab.tsx b/dashboard/src/features/jobs/components/job-errors-tab.tsx index 9418aa09..bd69fd39 100644 --- a/dashboard/src/features/jobs/components/job-errors-tab.tsx +++ b/dashboard/src/features/jobs/components/job-errors-tab.tsx @@ -1,5 +1,5 @@ import { AlertOctagon } from "lucide-react"; -import { EmptyState, ErrorState, Skeleton } from "@/components/ui"; +import { EmptyState, ErrorState, Skeleton, TaskErrorBlock } from "@/components/ui"; import type { JobError } from "@/lib/api-types"; import { formatAbsolute } from "@/lib/time"; @@ -44,9 +44,7 @@ export function JobErrorsTab({ errors, loading, error, onRetry }: JobErrorsTabPr {formatAbsolute(err.failed_at)}
- 
-            {err.error}
-
+ ))} diff --git a/dashboard/src/features/jobs/components/job-overview-tab.tsx b/dashboard/src/features/jobs/components/job-overview-tab.tsx index 25ca2be6..85c73785 100644 --- a/dashboard/src/features/jobs/components/job-overview-tab.tsx +++ b/dashboard/src/features/jobs/components/job-overview-tab.tsx @@ -1,6 +1,13 @@ import { ExternalLink as ExternalLinkIcon } from "lucide-react"; import { type ReactNode, useMemo } from "react"; -import { buttonVariants, Card, CardContent, CardHeader, CardTitle } from "@/components/ui"; +import { + buttonVariants, + Card, + CardContent, + CardHeader, + CardTitle, + TaskErrorBlock, +} from "@/components/ui"; import { applyJobContext, useIntegrations } from "@/features/settings"; import type { Job } from "@/lib/api-types"; import { cn } from "@/lib/cn"; @@ -99,9 +106,7 @@ export function JobOverviewTab({ job }: JobOverviewTabProps) { Last error - 
-              {job.error}
-
+ ) : null} diff --git a/dashboard/src/features/jobs/components/job-table.tsx b/dashboard/src/features/jobs/components/job-table.tsx index 8e2eecf7..274f5ba7 100644 --- a/dashboard/src/features/jobs/components/job-table.tsx +++ b/dashboard/src/features/jobs/components/job-table.tsx @@ -7,6 +7,7 @@ import { ErrorState, StatusBadge, TableSkeleton, + TaskErrorSummary, Tooltip, TooltipContent, TooltipTrigger, @@ -99,9 +100,10 @@ export function JobTable({ jobs, loading, error, onRetry }: JobTableProps) { const err = getValue(); if (!err) return —; return ( - - {err} - + ); }, }); diff --git a/dashboard/src/features/workflows/components/workflow-nodes-table.tsx b/dashboard/src/features/workflows/components/workflow-nodes-table.tsx index 27bc49ae..e5dc9c64 100644 --- a/dashboard/src/features/workflows/components/workflow-nodes-table.tsx +++ b/dashboard/src/features/workflows/components/workflow-nodes-table.tsx @@ -1,6 +1,6 @@ import { Link } from "@tanstack/react-router"; import { Layers } from "lucide-react"; -import { Badge, EmptyState } from "@/components/ui"; +import { Badge, EmptyState, TaskErrorSummary } from "@/components/ui"; import type { WorkflowNode } from "@/lib/api-types"; import { WORKFLOW_NODE_LABEL, WORKFLOW_NODE_TONE } from "@/lib/status"; import { formatDuration, formatRelative } from "@/lib/time"; @@ -76,7 +76,7 @@ function NodeRow({ node }: { node: WorkflowNode }) { {duration ?? "—"} - {node.error ?? ""} + {node.error ? : ""} ); diff --git a/dashboard/src/lib/task-error.test.ts b/dashboard/src/lib/task-error.test.ts new file mode 100644 index 00000000..6b6a665c --- /dev/null +++ b/dashboard/src/lib/task-error.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from "vitest"; +import { parseTaskError, taskErrorSummary, taskErrorTooltip } from "./task-error"; + +describe("parseTaskError", () => { + it("parses the canonical contract shape", () => { + expect( + parseTaskError( + '{"errtype":"BoomError","message":"it broke","traceback":["frame1","frame2"]}', + ), + ).toEqual({ + errtype: "BoomError", + message: "it broke", + traceback: ["frame1", "frame2"], + }); + }); + + it("accepts an empty message and empty traceback", () => { + expect(parseTaskError('{"errtype":"ValueError","message":"","traceback":[]}')).toEqual({ + errtype: "ValueError", + message: "", + traceback: [], + }); + }); + + it("tolerates surrounding whitespace", () => { + expect(parseTaskError(' {"errtype":"E","message":"m","traceback":[]}\n')).toEqual({ + errtype: "E", + message: "m", + traceback: [], + }); + }); + + it("returns null for legacy plain strings", () => { + expect(parseTaskError("ValueError: permanent failure")).toBeNull(); + expect( + parseTaskError('Traceback (most recent call last):\n File "x.py"\nValueError: boom'), + ).toBeNull(); + expect(parseTaskError("timed out after 30000ms")).toBeNull(); + }); + + it("returns null for JSON that is not an object", () => { + expect(parseTaskError('["errtype","message"]')).toBeNull(); + expect(parseTaskError('"just a string"')).toBeNull(); + expect(parseTaskError("42")).toBeNull(); + expect(parseTaskError("null")).toBeNull(); + }); + + it("returns null for objects without a string message", () => { + expect(parseTaskError('{"errtype":"E","traceback":[]}')).toBeNull(); + expect(parseTaskError('{"errtype":"E","message":42,"traceback":[]}')).toBeNull(); + expect(parseTaskError('{"errtype":"E","message":null}')).toBeNull(); + }); + + it("returns null for empty / null / undefined input", () => { + expect(parseTaskError("")).toBeNull(); + expect(parseTaskError(" ")).toBeNull(); + expect(parseTaskError(null)).toBeNull(); + expect(parseTaskError(undefined)).toBeNull(); + }); + + it("returns null for malformed JSON starting with a brace", () => { + expect(parseTaskError('{"errtype": broken')).toBeNull(); + }); + + it("defaults errtype when missing or malformed", () => { + expect(parseTaskError('{"message":"m"}')?.errtype).toBe("Error"); + expect(parseTaskError('{"errtype":7,"message":"m"}')?.errtype).toBe("Error"); + expect(parseTaskError('{"errtype":"","message":"m"}')?.errtype).toBe("Error"); + }); + + it("normalizes malformed tracebacks", () => { + expect(parseTaskError('{"errtype":"E","message":"m"}')?.traceback).toEqual([]); + expect(parseTaskError('{"errtype":"E","message":"m","traceback":"nope"}')?.traceback).toEqual( + [], + ); + expect( + parseTaskError('{"errtype":"E","message":"m","traceback":["a",1,"b",null]}')?.traceback, + ).toEqual(["a", "b"]); + }); +}); + +describe("taskErrorSummary", () => { + it("joins errtype and message", () => { + expect(taskErrorSummary({ errtype: "BoomError", message: "it broke", traceback: [] })).toBe( + "BoomError: it broke", + ); + }); + + it("omits the separator for an empty message", () => { + expect(taskErrorSummary({ errtype: "BoomError", message: "", traceback: [] })).toBe( + "BoomError", + ); + }); +}); + +describe("taskErrorTooltip", () => { + it("is just the summary when there is no traceback", () => { + expect(taskErrorTooltip({ errtype: "E", message: "m", traceback: [] })).toBe("E: m"); + }); + + it("appends the full traceback when short", () => { + expect(taskErrorTooltip({ errtype: "E", message: "m", traceback: ["f1", "f2"] })).toBe( + "E: m\n\nf1\nf2", + ); + }); + + it("keeps only the innermost frames of a long traceback", () => { + const traceback = ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]; + expect(taskErrorTooltip({ errtype: "E", message: "m", traceback })).toBe( + "E: m\n\n…\nf3\nf4\nf5\nf6\nf7", + ); + }); +}); diff --git a/dashboard/src/lib/task-error.ts b/dashboard/src/lib/task-error.ts new file mode 100644 index 00000000..30309bdf --- /dev/null +++ b/dashboard/src/lib/task-error.ts @@ -0,0 +1,62 @@ +/** + * Structured task error per the cross-SDK binding contract: workers report + * task failures as canonical JSON `{"errtype", "message", "traceback"}`. + * Anything that doesn't parse as a JSON object with a string `message` key + * (legacy strings, core-generated maintenance errors) must be shown as-is. + */ +export interface StructuredTaskError { + errtype: string; + message: string; + traceback: string[]; +} + +/** Innermost frames carry the signal; cap so native tooltips stay readable. */ +const TOOLTIP_TRACEBACK_FRAMES = 5; + +/** + * Parse an error string into its structured form, or `null` when it's a + * plain legacy string that must render verbatim (the contract's fallback + * rule: only a JSON object with a string `message` counts as structured). + */ +export function parseTaskError(raw: string | null | undefined): StructuredTaskError | null { + if (!raw) return null; + const trimmed = raw.trim(); + // Cheap reject before JSON.parse — most legacy errors (tracebacks, core + // maintenance strings) don't start with "{". + if (!trimmed.startsWith("{")) return null; + + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + return null; + } + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return null; + + const obj = parsed as Record; + if (typeof obj.message !== "string") return null; + + // `errtype`/`traceback` are required by the contract but tolerated when + // malformed — `message` alone decides structured-vs-legacy. + return { + errtype: typeof obj.errtype === "string" && obj.errtype ? obj.errtype : "Error", + message: obj.message, + traceback: Array.isArray(obj.traceback) + ? obj.traceback.filter((frame): frame is string => typeof frame === "string") + : [], + }; +} + +/** One-line headline: `errtype: message`, or just `errtype` when message is empty. */ +export function taskErrorSummary(error: StructuredTaskError): string { + return error.message ? `${error.errtype}: ${error.message}` : error.errtype; +} + +/** Tooltip text: the summary plus the tail of the traceback (innermost frames). */ +export function taskErrorTooltip(error: StructuredTaskError): string { + const summary = taskErrorSummary(error); + if (error.traceback.length === 0) return summary; + const frames = error.traceback.slice(-TOOLTIP_TRACEBACK_FRAMES); + const ellipsis = frames.length < error.traceback.length ? "…\n" : ""; + return `${summary}\n\n${ellipsis}${frames.join("\n")}`; +} diff --git a/dashboard/src/routes/workflows_.$id.tsx b/dashboard/src/routes/workflows_.$id.tsx index 9721dfad..86117858 100644 --- a/dashboard/src/routes/workflows_.$id.tsx +++ b/dashboard/src/routes/workflows_.$id.tsx @@ -22,6 +22,7 @@ import { workflowRunQuery, } from "@/features/workflows"; import { WORKFLOW_STATE_LABEL, WORKFLOW_STATE_TONE } from "@/lib/status"; +import { parseTaskError, taskErrorSummary } from "@/lib/task-error"; import { formatAbsolute, formatDuration } from "@/lib/time"; export const Route = createFileRoute("/workflows_/$id")({ @@ -89,7 +90,10 @@ function WorkflowDetailPage() { kvItems.push({ label: "Parent run", value: data.parent_run_id, mono: true }); } if (data.error) { - kvItems.push({ label: "Error", value: data.error }); + // Structured task errors collapse to their headline; the KV list is + // one-line-per-item and can't host a traceback block. + const structured = parseTaskError(data.error); + kvItems.push({ label: "Error", value: structured ? taskErrorSummary(structured) : data.error }); } return ( From e51369d4ef05da1f7a2a77350d903508579c41e0 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:59:54 +0530 Subject: [PATCH 6/8] fix(node): derive errtype from the runtime class name --- sdks/node/src/task-error.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sdks/node/src/task-error.ts b/sdks/node/src/task-error.ts index 63de1d96..224efd0f 100644 --- a/sdks/node/src/task-error.ts +++ b/sdks/node/src/task-error.ts @@ -20,7 +20,9 @@ export function encodeTaskError(error: unknown): string { let message: string; let traceback: string[] = []; if (error instanceof Error) { - errtype = error.name || error.constructor.name || "Error"; + // Prefer the runtime class name: `error.name` stays "Error" for subclasses + // that don't override it, collapsing distinct types to the generic value. + errtype = error.constructor?.name || error.name || "Error"; message = error.message ?? String(error); // Keep stack lines verbatim — readers render them as-is. traceback = error.stack ? error.stack.split("\n") : []; From 0e7614f6f96a90d27e4c99319f0e1ecf8661e84a Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:59:57 +0530 Subject: [PATCH 7/8] fix(java): default missing errtype to Error to match contract --- .../src/main/java/org/byteveda/taskito/errors/TaskError.java | 3 +++ .../main/java/org/byteveda/taskito/errors/TaskErrors.java | 5 ++++- .../java/org/byteveda/taskito/errors/TaskErrorsTest.java | 4 +++- 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java index c14c1f01..bfba7811 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskError.java @@ -28,4 +28,7 @@ public final class TaskError { public String summary() { return errtype.isEmpty() ? message : errtype + ": " + message; } + + // errtype is empty only when a caller constructs a TaskError directly with + // an empty type; decode() defaults a missing wire errtype to "Error". } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java index f13baf26..b37c6066 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/TaskErrors.java @@ -52,8 +52,11 @@ public static TaskError decode(String raw) { if (message == null || !message.isTextual()) { return null; } + // Default a missing errtype to "Error" to match the cross-SDK contract + // (other SDKs decode it identically), so the same JSON summarizes the + // same way everywhere. JsonNode errtype = node.get("errtype"); - String errtypeText = errtype != null && errtype.isTextual() ? errtype.asText() : ""; + String errtypeText = errtype != null && errtype.isTextual() ? errtype.asText() : "Error"; return new TaskError(errtypeText, message.asText(), readTraceback(node.get("traceback")), raw); } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java b/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java index 65690ece..fab662f0 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/errors/TaskErrorsTest.java @@ -75,7 +75,9 @@ void summarizePrefersErrtypeMessageHeadline() { assertEquals( "BoomError: it broke", TaskErrors.summarize("{\"errtype\":\"BoomError\",\"message\":\"it broke\",\"traceback\":[]}")); - assertEquals("just text", TaskErrors.summarize("{\"message\":\"just text\"}")); // no errtype + // A missing errtype defaults to "Error" (cross-SDK contract), so the + // headline is "Error: ", matching every other SDK's summary. + assertEquals("Error: just text", TaskErrors.summarize("{\"message\":\"just text\"}")); assertEquals( "task timed out after 5000ms", TaskErrors.summarize("task timed out after 5000ms")); // raw fallback } From 62cfc1bc7ee602f9781a47d7d9f3257ec988eb80 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:59:57 +0530 Subject: [PATCH 8/8] docs(python): correct JobErrorDetails field-default docstring --- sdks/python/taskito/exceptions.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/sdks/python/taskito/exceptions.py b/sdks/python/taskito/exceptions.py index ded8a154..c489f43d 100644 --- a/sdks/python/taskito/exceptions.py +++ b/sdks/python/taskito/exceptions.py @@ -20,9 +20,10 @@ class TaskCancelledError(TaskitoError): class JobErrorDetails: """Structured failure details mixin for job-outcome exceptions. - Populated when the stored error is a canonical structured task error - (see ``taskito.task_errors``); all fields default to ``None`` for - legacy/system plain-string errors. + ``job_id`` and ``raw_error`` are always populated. ``errtype`` and + ``traceback`` carry the decoded canonical task error (see + ``taskito.task_errors``) and are ``None`` when the stored error is a + legacy/system plain string. """ errtype: str | None = None