From aeaea57ca1a0fac02157035530e84c3588a7aec1 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 28 Jul 2026 13:15:24 +0200 Subject: [PATCH] fix(wac): one failure record for tasks and steps, in every round (#10368) * fix(wac): hand a caught task and step failure the same shape in every round Co-Authored-By: Claude Opus 5 (1M context) * refactor(wac): decide the failure record once, server-side Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): leave a legacy SDK's failure marker untouched, and ship wacError to jsr Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): carry a step's custom error fields, and bound the stack in bytes Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): keep a step's extra fields serializable and bounded Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): record a non-Error throw the way a task records it Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): guard the last unguarded throw site in the step marker Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): make failure reporting non-throwing on both clients Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): take the step traceback the way the executor takes it Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): contain the reads that happen before a failure is checkpointed Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): fall back to the checkpointed marker, not the live one Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): keep non-finite fields and hostile proxies out of the checkpoint path Co-Authored-By: Claude Opus 5 (1M context) * fix(wac): keep the snapshot that passed the serialization probe Co-Authored-By: Claude Opus 5 (1M context) * chore(wac): keep the failure-record module's surface to what is used Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- backend/windmill-api/src/jobs.rs | 17 +- backend/windmill-common/src/wac.rs | 339 +++++++++++++++++- .../windmill-worker/src/result_processor.rs | 12 +- cli/src/guidance/skills.gen.ts | 15 +- python-client/wmill/tests/test_workflow.py | 199 +++++++++- python-client/wmill/wmill/client.py | 173 +++++++-- system_prompts/auto-generated/prompts.ts | 15 +- .../auto-generated/sdks/wac-python.md | 13 +- .../skills/write-workflow-as-code/SKILL.md | 15 +- system_prompts/base/workflow-as-code.md | 2 + system_prompts/generate.py | 6 + typescript-client/build.jsr.sh | 1 + typescript-client/build.sh | 1 + typescript-client/client.ts | 82 ++--- typescript-client/jsr.json | 2 +- typescript-client/tests/workflow.test.ts | 228 ++++++++++-- typescript-client/wacError.ts | 159 ++++++++ 17 files changed, 1143 insertions(+), 136 deletions(-) create mode 100644 typescript-client/wacError.ts diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index f030aaa3d1..663f65ca6f 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -5812,6 +5812,15 @@ pub struct WacInlineCheckpointPayload { pub duration_ms: Option, } +#[derive(Serialize)] +pub struct WacInlineCheckpointResponse { + /// The normalized failure record, when the posted step failed. The SDK + /// raises from this rather than from its own copy, so the round that ran + /// the failing body and every replay of it read the same object. + #[serde(skip_serializing_if = "Option::is_none")] + pub failure: Option, +} + /// Fast-path endpoint called by the WAC v2 SDKs to persist a single `step()` /// checkpoint delta without unwinding the parent workflow subprocess. /// @@ -5837,7 +5846,7 @@ pub async fn wac_inline_checkpoint( Extension(db): Extension, Path((w_id, job_id)): Path<(String, Uuid)>, Json(payload): Json, -) -> error::Result { +) -> error::Result> { // Enforce ephemeral-job-token binding: the presented token must be the // one issued to *this* specific job. Regular workspace API tokens don't // have `job_id` populated in their JWT claims, so `token_job_id` is None @@ -5869,7 +5878,7 @@ pub async fn wac_inline_checkpoint( let source_hash = runnable_id.map(|h| h.to_string()); let mut tx = db.begin().await?; - windmill_common::wac::persist_inline_checkpoint_delta( + let failure = windmill_common::wac::persist_inline_checkpoint_delta( &mut tx, &job_id, source_hash.as_deref(), @@ -5881,7 +5890,9 @@ pub async fn wac_inline_checkpoint( .await?; tx.commit().await?; - Ok(StatusCode::OK) + // Only failures come back: a successful step's result can be large, and the + // SDK already holds it. + Ok(Json(WacInlineCheckpointResponse { failure })) } lazy_static::lazy_static! { diff --git a/backend/windmill-common/src/wac.rs b/backend/windmill-common/src/wac.rs index 51abc5d60f..d46c937e33 100644 --- a/backend/windmill-common/src/wac.rs +++ b/backend/windmill-common/src/wac.rs @@ -80,6 +80,172 @@ mod tests { assert_eq!(approval_resume_id("approval_2"), 0x50d1_eeca); assert_eq!(approval_resume_id("manager"), 0x6ee4_a469); } + + use super::wac_failure_record; + use serde_json::json; + + /// The point of the function: a task failure and a step failure describing + /// the same error must be indistinguishable to the handler that catches + /// them, apart from the child job only a task has. + #[test] + fn a_task_and_a_step_failure_read_the_same() { + let from_child = wac_failure_record( + "fetch", + Some("abc-123"), + &json!({"error": {"name": "ValueError", "message": "nope", "stack": "frames"}}), + ); + let from_step = wac_failure_record( + "fetch", + None, + &json!({"error": {"name": "ValueError", "message": "nope", "stack": "frames"}}), + ); + assert_eq!(from_child["result"], from_step["result"]); + assert_eq!(from_child["message"], json!("nope")); + assert_eq!(from_step["message"], json!("nope")); + assert_eq!(from_child["child_job_id"], json!("abc-123")); + assert_eq!(from_step.get("child_job_id"), None); + } + + /// A child job's result is whatever the failing job produced — a cancel, a + /// timeout, an executor that writes a bare string. The handler is still + /// promised `name` and `message`, so they cannot be conjured per-caller. + #[test] + fn an_unshaped_child_result_still_yields_name_and_message() { + for raw in [ + json!({"error": "boom"}), + json!({"error": {"message": "boom"}}), + json!("boom"), + json!(null), + ] { + let rec = wac_failure_record("s", None, &raw); + assert_eq!(rec["result"]["error"]["name"], json!("Error"), "{raw}"); + assert!( + rec["result"]["error"]["message"].is_string(), + "{raw} produced no message" + ); + assert_eq!(rec["result"]["error"].get("stack"), None, "{raw}"); + } + // Nothing to say beats saying `{}`: the fallback exists for these. + for empty in [json!(null), json!({}), json!([]), json!({"error": ""})] { + assert_eq!( + wac_failure_record("s", None, &empty)["message"], + json!("WAC step 's' failed"), + "{empty}" + ); + } + } + + /// Extra keys are the failing side's own, and dropping them would lose a + /// custom error's fields; the three normalized ones still win. + #[test] + fn extra_error_fields_survive_normalization() { + let rec = wac_failure_record( + "s", + None, + &json!({"error": {"name": "HttpError", "message": "429", "code": 429, "stack": 12}}), + ); + assert_eq!(rec["result"]["error"]["code"], json!(429)); + assert_eq!(rec["result"]["error"]["name"], json!("HttpError")); + // a non-string stack is not something a handler can be told to read + assert_eq!(rec["result"]["error"].get("stack"), None); + } + + use super::normalize_posted_step_result; + + /// An SDK that predates the echoed record raises from the copy it built, so + /// rewriting what it posted would make its live round and its replays + /// disagree — the very thing being fixed here. It keeps its own shape. + #[test] + fn a_legacy_sdk_marker_is_stored_untouched() { + let legacy = json!({ + "__wmill_error": true, + "message": "nope", + "step_key": "s", + "result": {"error": "nope", "type": "TypeError"}, + }); + assert_eq!(normalize_posted_step_result("s", legacy.clone()), legacy); + } + + #[test] + fn a_current_sdk_marker_is_normalized_and_a_success_is_not() { + let posted = json!({ + "__wmill_error": true, + "message": "nope", + "step_key": "s", + "result": {"error": {"name": "ValueError", "message": "nope"}}, + }); + let stored = normalize_posted_step_result("s", posted); + assert_eq!(stored["result"]["error"]["name"], json!("ValueError")); + assert_eq!(stored["step_key"], json!("s")); + + let success = json!({"rows": 3}); + assert_eq!(normalize_posted_step_result("s", success.clone()), success); + } + + #[test] + fn an_oversized_stack_is_truncated() { + let rec = wac_failure_record( + "s", + None, + &json!({"error": {"message": "m", "stack": "x".repeat(100_000)}}), + ); + let stack = rec["result"]["error"]["stack"].as_str().unwrap(); + assert!(stack.len() < 100_000, "stack was not truncated"); + assert!(stack.ends_with("... (truncated)")); + } + + /// The cap bounds what lands in the checkpoint, so it has to be bytes: a + /// multibyte traceback counted in characters would be up to 4x over. + #[test] + fn the_stack_cap_counts_bytes_not_characters() { + let rec = wac_failure_record( + "s", + None, + &json!({"error": {"message": "m", "stack": "é".repeat(50_000)}}), + ); + let stack = rec["result"]["error"]["stack"].as_str().unwrap(); + assert!( + stack.len() <= 8 * 1024 + "\n... (truncated)".len(), + "kept {} bytes", + stack.len() + ); + } + + /// `extra` is the failing side's own attributes, so it can carry a response + /// body straight past the cap that exists to bound the checkpoint. + #[test] + fn an_oversized_extra_is_dropped_rather_than_stored() { + let rec = wac_failure_record( + "s", + None, + &json!({"error": {"message": "m", "extra": {"body": "x".repeat(100_000)}}}), + ); + assert_eq!(rec["result"]["error"].get("extra"), None); + assert_eq!(rec["result"]["error"]["extra_omitted"], json!(true)); + + // one that fits is kept whole + let small = wac_failure_record( + "s", + None, + &json!({"error": {"message": "m", "extra": {"code": 429}}}), + ); + assert_eq!(small["result"]["error"]["extra"], json!({"code": 429})); + assert_eq!(small["result"]["error"].get("extra_omitted"), None); + } + + /// `message` is the failure's own message; `Value::to_string` on a string + /// would hand the handler `"boom"` with the JSON quotes still on it. + #[test] + fn a_bare_string_failure_keeps_its_message_unquoted() { + assert_eq!( + wac_failure_record("s", None, &json!({"error": "boom"}))["message"], + json!("boom") + ); + assert_eq!( + wac_failure_record("s", None, &json!("boom"))["message"], + json!("boom") + ); + } } /// Load the WAC checkpoint from `v2_job_status.workflow_as_code_status._checkpoint`. @@ -139,6 +305,164 @@ pub async fn save_checkpoint( Ok(()) } +/// Marks a `completed_steps` entry as a failure rather than a step result. +pub(crate) const WAC_ERROR_MARKER: &str = "__wmill_error"; + +/// Per-field budget for the two unbounded things a failure record carries, the +/// stack and `extra`. `persist_inline_checkpoint_delta` rewrites the whole +/// checkpoint on every step, so either one left unbounded is re-serialized once +/// per subsequent step for the rest of the workflow. The two are additive, so a +/// record costs at most twice this. +const MAX_CHECKPOINT_FIELD_BYTES: usize = 8 * 1024; + +fn truncate_stack(stack: &str) -> String { + if stack.len() <= MAX_CHECKPOINT_FIELD_BYTES { + return stack.to_string(); + } + // Byte budget, not characters: the cap exists to bound what goes into the + // checkpoint, and a multibyte traceback would otherwise be up to 4x it. + let mut cut = MAX_CHECKPOINT_FIELD_BYTES; + while cut > 0 && !stack.is_char_boundary(cut) { + cut -= 1; + } + format!("{}\n... (truncated)", &stack[..cut]) +} + +/// A failure's own message, without the JSON quoting `Value::to_string` puts +/// around a string. `None` when the value carries no message at all, so the +/// caller's fallback wins — a reader handed `{}` learns less than one handed +/// "WAC step 'x' failed". +fn value_message(v: &Value) -> Option { + match v { + Value::Null => None, + Value::String(s) if s.is_empty() => None, + Value::String(s) => Some(s.clone()), + Value::Object(o) if o.is_empty() => None, + Value::Array(a) if a.is_empty() => None, + other => Some(other.to_string()), + } +} + +fn message_only(message: Option) -> serde_json::Map { + let mut m = serde_json::Map::new(); + if let Some(message) = message { + m.insert("message".to_string(), Value::String(message)); + } + m +} + +/// Build the failure record a caught WAC failure reads, from whatever the +/// failing side produced. +/// +/// The single place this shape is decided, for both a task failure (arriving as +/// the child job's own result) and a `step()` failure (as the SDK posted it). +/// Assembling it per caller instead is how the two come to disagree on `name` +/// or on `stack` while both claim to be one shape. `name`, `message` and +/// `stack` are normalized here; any other key the failing side attached to its +/// error is passed through untouched, so a custom error's own fields survive. +pub fn wac_failure_record(step_key: &str, child_job_id: Option<&str>, raw_result: &Value) -> Value { + let mut error = match raw_result.get("error") { + Some(Value::Object(o)) => o.clone(), + // A bare-string error (some executors), or a result not shaped like a + // failure at all: keep whatever it says as the message rather than + // dropping it. + Some(other) => message_only(value_message(other)), + None => message_only(value_message(raw_result)), + }; + + let name = error + .get("name") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .unwrap_or("Error") + .to_string(); + let message = error + .get("message") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()) + .unwrap_or_else(|| format!("WAC step '{step_key}' failed")); + error.insert("name".to_string(), Value::String(name)); + error.insert("message".to_string(), Value::String(message.clone())); + match error.get("stack").and_then(|v| v.as_str()) { + Some(stack) => { + error.insert("stack".to_string(), Value::String(truncate_stack(stack))); + } + // Never invent one, and never keep a non-string in the field a handler + // is told it can read. + None => { + error.remove("stack"); + } + } + + // `extra` is the failing side's own attributes, so it can hold a response + // body or a dataframe repr and route straight around the stack cap into the + // checkpoint this record is rewritten into on every later step. Dropped + // wholesale past the same budget rather than truncated, since half a + // structure is worse than a flag saying it was too big. + if let Some(extra) = error.get("extra") { + if serde_json::to_string(extra).map_or(true, |s| s.len() > MAX_CHECKPOINT_FIELD_BYTES) { + error.remove("extra"); + error.insert("extra_omitted".to_string(), Value::Bool(true)); + } + } + + let mut record = serde_json::Map::new(); + record.insert(WAC_ERROR_MARKER.to_string(), Value::Bool(true)); + // `str(e)` / `e.message` reads the failure's own message whether it came + // from a task or a step; which task, and which child job, are the fields + // below rather than prose baked into the message. + record.insert("message".to_string(), Value::String(message)); + record.insert("step_key".to_string(), Value::String(step_key.to_string())); + if let Some(child) = child_job_id { + record.insert("child_job_id".to_string(), Value::String(child.to_string())); + } + record.insert( + "result".to_string(), + Value::Object( + [("error".to_string(), Value::Object(error))] + .into_iter() + .collect(), + ), + ); + Value::Object(record) +} + +/// Decide what to store for a step result an SDK posted. +/// +/// A failure is normalized through `wac_failure_record`, the same function that +/// shapes task failures, so the two cannot drift apart. +/// +/// A marker an older SDK posted is stored untouched instead. Those are +/// recognizable by an `error` that is a message string rather than an object, +/// and the SDK that posted one raises from the copy it built and ignores the +/// record echoed back to it. Rewriting it here would leave the round that ran +/// the failing body reading one shape and every replay of it reading another — +/// the divergence this whole mechanism exists to remove. It keeps its own shape +/// until it upgrades. +pub(crate) fn normalize_posted_step_result(key: &str, posted: Value) -> Value { + if !is_wac_failure(&posted) { + return posted; + } + let normalizable = posted + .get("result") + .and_then(|r| r.get("error")) + .map(|e| e.is_object()) + .unwrap_or(false); + if !normalizable { + return posted; + } + wac_failure_record(key, None, posted.get("result").unwrap_or(&Value::Null)) +} + +/// Whether a `completed_steps` entry is a failure record. +pub(crate) fn is_wac_failure(value: &Value) -> bool { + value + .get(WAC_ERROR_MARKER) + .and_then(|v| v.as_bool()) + .unwrap_or(false) +} + /// Process a completed child job result: add to checkpoint's completed_steps. pub fn add_completed_step(checkpoint: &mut WacCheckpoint, step_key: &str, result: Value) { checkpoint @@ -225,6 +549,11 @@ pub fn add_completed_step(checkpoint: &mut WacCheckpoint, step_key: &str, result /// multiple times per call, and the `||` merges re-serialize the whole /// object. The two-statement Rust-side load-modify-save below is ~10× /// faster in practice, so we keep it and rely on the SDK-level lock. +/// +/// Returns the value actually stored: a failure posted by an SDK is normalized +/// through `wac_failure_record` first, so the round that ran the failing body +/// can raise from the same record every replay will read instead of building +/// its own copy of it. pub async fn persist_inline_checkpoint_delta( tx: &mut Transaction<'_, Postgres>, job_id: &Uuid, @@ -233,7 +562,7 @@ pub async fn persist_inline_checkpoint_delta( result: Value, started_at: Option<&str>, duration_ms: Option, -) -> error::Result<()> { +) -> error::Result> { // Row-lock the existing checkpoint row (if any) for the duration of the // transaction. NULL if the row doesn't exist yet — see the doc comment // above for why the first-write race is accepted. @@ -288,6 +617,12 @@ pub async fn persist_inline_checkpoint_delta( "WAC v2 inline checkpoint — persisting step result" ); + let result = normalize_posted_step_result(key, result); + + // Only a failure is ever read back, so only a failure is copied: a + // successful step's result can be large and moves straight into the + // checkpoint. + let failure = is_wac_failure(&result).then(|| result.clone()); add_completed_step(&mut checkpoint, key, result); let status_json = serde_json::to_value(&checkpoint) @@ -335,5 +670,5 @@ pub async fn persist_inline_checkpoint_delta( .await .map_err(|e| Error::InternalErr(format!("Failed to write step timeline: {e}")))?; - Ok(()) + Ok(failure) } diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index c95bd70899..1709e4a150 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -1791,13 +1791,11 @@ pub(crate) async fn handle_wac_child_completion( step_key = %step_key, "WAC v2 child job failed, storing error for workflow try/catch" ); - json!({ - "__wmill_error": true, - "message": format!("WAC task '{}' failed (child job {})", step_key, child_job_id), - "child_job_id": child_job_id.to_string(), - "step_key": step_key, - "result": child_err, - }) + windmill_common::wac::wac_failure_record( + &step_key, + Some(&child_job_id.to_string()), + &child_err, + ) }; tracing::info!( diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 5f1c5ab994..57a900f50e 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -6506,6 +6506,8 @@ Python: \`except Exception\` is safe around WAC calls because internal suspensio TypeScript: avoid broad \`try/catch\` around WAC SDK calls. The SDK uses an internal suspension error during initial dispatch; catching it can break workflow suspension. If a broad catch is unavoidable, rethrow internal suspension errors before handling business errors. +A caught failure reads the same whether it came from a task or from a \`step()\`, and the same in the round that ran the failing body as in every round replaying it. It carries \`step_key\`, \`child_job_id\` (absent for a \`step()\`, which runs in the workflow job and has no child job), a \`message\` that is the failure's own message, and \`result\` = \`{"error": {"name", "message", "stack"?, "extra"?}}\`. \`name\`, \`message\` and \`stack\` are the fields that read the same whichever side failed; \`name\` and \`message\` are always there, \`stack\` only when the failure had a traceback to give. \`extra\` carries the failure's own custom fields (an exception's attributes, an error's properties) and is best-effort: it is absent when there were none, and a task can report entries a step does not, so read it defensively and don't branch on its absence. \`extra\` is dropped when it is too large to keep in the checkpoint, and \`extra_omitted: true\` says so — absent \`extra\` with no \`extra_omitted\` means the failure simply had no custom fields. Branch on those, not on the original exception type: the workflow body re-runs from the top every round and a replay rebuilds the failure from the checkpoint, so nothing outside that record survives. Python raises \`TaskError\`; TypeScript throws an \`Error\` named \`TaskError\` carrying the same fields. Nothing is chained onto \`__cause__\` / \`cause\` — the traceback is in \`result.error.stack\`, and is also printed to the job log when the step fails. + ## TypeScript Workflow-as-Code API (windmill-client) @@ -6644,14 +6646,19 @@ export async function parallel(items: T[], fn: (item: T) => PromiseLike Import: \`from wmill import workflow, task, task_script, task_flow, step, sleep, wait_for_approval, get_approval_urls, get_resume_urls, parallel, TaskError\` \`\`\`python -# Raised when a WAC task step failed. +# Raised when a WAC \`\`task\`\` or \`\`step\`\` failed. # # Attributes: # step_key: The checkpoint key of the failed step. -# child_job_id: The UUID of the failed child job. -# result: The error result from the child job. +# child_job_id: The UUID of the failed child job, or \`\`None\`\` for a +# \`\`step()\`\`, which runs in the workflow job and has no child job. +# result: \`\`{"error": {"name", "message", "stack"?, "extra"?}}\`\` — the +# same shape whether a task or a step failed. \`\`name\`\` and \`\`message\`\` +# are always present; \`\`stack\`\` only when the failure had a traceback, +# and \`\`extra\`\` only when it carried custom fields of its own, dropped +# with \`\`extra_omitted: True\`\` beside it when too large to checkpoint. class TaskError(Exception): - def __init__(self, message: str, *, step_key: str = '', child_job_id: str = '', result = None) + def __init__(self, message: str, *, step_key: str = '', child_job_id: Optional[str] = None, result = None) # Get URLs needed for resuming a flow after suspension. # diff --git a/python-client/wmill/tests/test_workflow.py b/python-client/wmill/tests/test_workflow.py index 1cca524237..3f80fb06b3 100644 --- a/python-client/wmill/tests/test_workflow.py +++ b/python-client/wmill/tests/test_workflow.py @@ -976,13 +976,21 @@ class TestRaisingInlineStepIsCheckpointed: on the never-resolving future forever. """ + # A failed child job reports `{"error": {"name", "message", "stack"}}`, and a + # failed step() has to be indistinguishable from it. The stack is a traceback + # string, asserted separately. MARKER = { "__wmill_error": True, "message": "boom", "step_key": "risky", - "result": {"error": "boom", "type": "ValueError"}, + "result": {"error": {"name": "ValueError", "message": "boom"}}, } + @staticmethod + def _without_stack(marker: dict) -> dict: + error = {k: v for k, v in marker["result"]["error"].items() if k != "stack"} + return {**marker, "result": {**marker["result"], "error": error}} + @staticmethod def _boom(): raise ValueError("boom") @@ -1003,36 +1011,205 @@ class TestRaisingInlineStepIsCheckpointed: r = _run_workflow(self._wf(), {}, {"x": 5}) assert r["type"] == "inline_checkpoint" assert r["key"] == "risky" - assert r["result"] == self.MARKER + assert self._without_stack(r["result"]) == self.MARKER + stack = r["result"]["result"]["error"]["stack"] + # Frames only and the SDK's own `result = fn()` frame dropped, the way + # the python executor formats a failed job's stack. + assert 'raise ValueError("boom")' in stack + assert "result = fn()" not in stack + assert not stack.startswith("Traceback") + + def test_custom_exception_attributes_survive_under_extra(self): + """A failed child job reports custom attributes under ``error.extra``; + a step dropping them would make the same exception carry less depending + on how it was run.""" + from wmill.client import _step_error_marker + + class HttpError(ValueError): + def __init__(self): + super().__init__("429") + self.code = 429 + + error = _step_error_marker("k", HttpError())["result"]["error"] + assert error["extra"] == {"code": 429} + assert error["name"] == "HttpError" + assert "extra" not in _step_error_marker("k", ValueError("plain"))["result"]["error"] + + def test_unserializable_attributes_do_not_cost_the_fast_path(self): + """The fast-path POST serializes strictly, so an exception holding a + live object — ``resp.raise_for_status()`` is the common one — must not + make the marker unserializable and drop the step onto the slow path.""" + import json as _json + + from wmill.client import _step_error_marker + + class Boom(Exception): + def __init__(self): + super().__init__("boom") + self.response = object() + self.status = 429 + + marker = _step_error_marker("k", Boom()) + _json.dumps(marker) # raises if an attribute leaked through unserialized + assert marker["result"]["error"]["extra"]["status"] == 429 + + def test_an_exception_whose_str_raises_still_reports(self): + """Every coercion of the user's exception runs inside the ``except`` + reporting it, so one that raises would replace their failure with an + unrelated one and leave the step uncheckpointed.""" + import json as _json + + from wmill.client import _step_error_marker + + class Hostile(Exception): + def __str__(self): + raise RuntimeError("cannot be rendered") + + marker = _step_error_marker("k", Hostile()) + _json.dumps(marker) + assert marker["result"]["error"]["name"] == "Hostile" + assert "unrepresentable" in marker["result"]["error"]["message"] + + # ...including one that makes reading its own traceback raise + class HostileTraceback(Exception): + def __getattribute__(self, item): + if item == "__traceback__": + raise RuntimeError("no traceback for you") + return super().__getattribute__(item) + + marker = _step_error_marker("k", HostileTraceback()) + _json.dumps(marker) + assert marker["result"]["error"]["name"] == "HostileTraceback" + + # ...or reading its own attributes + class HostileDict(Exception): + def __getattribute__(self, item): + if item == "__dict__": + raise RuntimeError("no attributes for you") + return super().__getattribute__(item) + + marker = _step_error_marker("k", HostileDict()) + _json.dumps(marker) + assert marker["result"]["error"]["name"] == "HostileDict" + + # A float is serializable, so `default=` never sees NaN — it would go out + # as a bare `NaN` literal, which is not JSON and which the backend + # rejects, so the step could not be checkpointed at all. + class NotFinite(Exception): + def __init__(self): + super().__init__("nan") + self.value = float("nan") + self.limit = float("inf") + + marker = _step_error_marker("k", NotFinite()) + _json.dumps(marker, allow_nan=False) + assert marker["result"]["error"]["extra"] == {"value": "NaN", "limit": "Infinity"} def test_fast_path_posts_error_and_raises_the_replay_exception(self, monkeypatch): """The default path: the checkpoint is POSTed and the workflow body gets the same ``TaskError`` a replay rebuilds from the marker — raising the original ``ValueError`` here would make ``except ValueError:`` catch on - this run and miss on the next one.""" + this run and miss on the next one. ``except`` is control flow, so + anything a handler can branch on has to be identical in both rounds.""" _set_inline_fast_path_env(monkeypatch) - stub = _StubInlineClient() + class _EchoingStub(_StubInlineClient): + """The endpoint normalizes the failure before storing it and echoes + back what it stored. The echo deliberately differs from what was + posted, so the assertions below can tell which copy was raised from.""" + + async def post(self, url, content=None): + await super().post(url, content=content) + stored = {**self.posted[-1]["result"], "message": "normalized by the backend"} + + class _Response: + # the endpoint answers with a JSON body; a backend predating + # the echo answers without one, which is how the client tells + # "no echo" from "an echo it could not read" + headers = {"content-type": "application/json"} + + def raise_for_status(self): + pass + + def json(self): + return {"failure": stored} + + return _Response() + + stub = _EchoingStub() posted = stub.posted async def run(): ctx = WorkflowCtx({}) ctx._inline_http_client = stub - with pytest.raises(TaskError, match="boom") as live: + with pytest.raises(TaskError) as live: await ctx._run_inline_step("risky", self._boom) - # ...and the replay of that very checkpoint raises the same thing. - replayed = WorkflowCtx({"completed_steps": {"risky": self.MARKER}}) - with pytest.raises(TaskError, match="boom") as replay: + # The live round raised from the record the backend stored, not from + # the marker it posted: that is what keeps the two rounds identical + # even if the SDK and the backend ever build a record differently. + stored = {**posted[0]["result"], "message": "normalized by the backend"} + assert str(live.value) == "normalized by the backend" + + # ...and the replay of that very record raises the same thing. + replayed = WorkflowCtx({"completed_steps": {"risky": stored}}) + with pytest.raises(TaskError) as replay: await replayed._run_inline_step("risky", self._boom) assert type(live.value) is type(replay.value) assert live.value.args == replay.value.args - assert live.value.result == replay.value.result == self.MARKER["result"] - assert isinstance(live.value.__cause__, ValueError) + assert live.value.result == replay.value.result == stored["result"] + assert live.value.step_key == replay.value.step_key == "risky" + # A step has no child job to name, and nothing hangs off __cause__: + # a replay has no original exception to chain, so neither round does. + assert live.value.child_job_id is replay.value.child_job_id is None + assert live.value.__cause__ is replay.value.__cause__ is None asyncio.run(run()) assert len(posted) == 1 assert posted[0]["key"] == "risky" - assert posted[0]["result"] == self.MARKER + assert self._without_stack(posted[0]["result"]) == self.MARKER + + def test_a_missing_echo_is_not_the_same_as_an_unreadable_one(self, monkeypatch): + """A backend predating the echo answers without a JSON body and the + locally checkpointed marker stands in. A JSON body that will not parse + means the stored record exists but is unknown, so the round has to end + and let the next one read whatever the backend actually kept.""" + _set_inline_fast_path_env(monkeypatch) + + def _client(headers, json_impl): + class _Response: + def raise_for_status(self): + pass + + _Response.headers = headers + _Response.json = json_impl + + class _Client(_StubInlineClient): + async def post(self, url, content=None): + await super().post(url, content=content) + return _Response() + + return _Client() + + def _boom_json(self): + raise ValueError("not json") + + async def run(): + # no JSON body: the fast path still completes, raising the failure + ctx = WorkflowCtx({}) + ctx._inline_http_client = _client({}, _boom_json) + with pytest.raises(TaskError): + await ctx._run_inline_step("risky", self._boom) + + # a JSON body that will not parse: fall through to the suspend path + ctx = WorkflowCtx({}) + ctx._inline_http_client = _client( + {"content-type": "application/json"}, _boom_json + ) + with pytest.raises(_StepSuspend) as suspend: + await ctx._run_inline_step("risky", self._boom) + assert suspend.value.dispatch_info["key"] == "risky" + + asyncio.run(run()) def test_replay_reraises_and_does_not_hang(self): checkpoint = { diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 682a2c4c9c..05496f0724 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -2654,6 +2654,8 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]: import asyncio as _asyncio import contextvars as _contextvars +import sys as _sys +import traceback as _traceback def _assert_usable_step_key(key: str, what: str) -> None: @@ -2687,21 +2689,62 @@ class _StepFailure(BaseException): class TaskError(Exception): - """Raised when a WAC task step failed. + """Raised when a WAC ``task`` or ``step`` failed. Attributes: step_key: The checkpoint key of the failed step. - child_job_id: The UUID of the failed child job. - result: The error result from the child job. + child_job_id: The UUID of the failed child job, or ``None`` for a + ``step()``, which runs in the workflow job and has no child job. + result: ``{"error": {"name", "message", "stack"?, "extra"?}}`` — the + same shape whether a task or a step failed. ``name`` and ``message`` + are always present; ``stack`` only when the failure had a traceback, + and ``extra`` only when it carried custom fields of its own, dropped + with ``extra_omitted: True`` beside it when too large to checkpoint. """ - def __init__(self, message: str, *, step_key: str = "", child_job_id: str = "", result=None): + def __init__(self, message: str, *, step_key: str = "", child_job_id: Optional[str] = None, result=None): super().__init__(message) self.step_key = step_key self.child_job_id = child_job_id self.result = result +def _safe_str(o) -> str: + """``str()`` on the failing side's own object, which can raise in turn — a + detached ORM row, a proxy over a closed connection, an ``__str__`` that + itself fails. Every coercion here runs inside the ``except`` that is + reporting the user's failure, so an escape would replace their error with an + unrelated one and skip the checkpoint entirely.""" + try: + return str(o) + except Exception: + return f"" + + +def _step_error_stack(exc: BaseException) -> str: + """The traceback of a failed ``step()`` body, formatted the way the python + executor formats a failed job's: frames only, and the frame that called into + the user's code dropped. Here that first frame is ``_run_inline_step``'s own + ``result = fn()``, the counterpart of the generated wrapper frame the + executor strips, so a step's stack and a task's stack read alike. + + Taken from ``sys.exc_info()`` the way the executor takes it, falling back to + the attribute: an exception overriding ``__getattribute__`` makes reading + ``__traceback__`` raise, and this runs inside the ``except`` reporting the + user's failure, so an escape would lose both their error and the checkpoint. + """ + tb = _sys.exc_info()[2] + if tb is None: + try: + tb = exc.__traceback__ + except Exception: + return "" + try: + return "".join(_traceback.format_tb(tb)[1:]).strip() + except Exception: + return "" + + def _json_round_trip(value): """Put a value through the checkpoint's encoding without checkpointing it, so the paths that never persist anything still hand back the shape the ones that @@ -2711,23 +2754,69 @@ def _json_round_trip(value): def _step_error_marker(key: str, exc: BaseException) -> dict: """Serialize a failed ``step()`` body into the ``__wmill_error`` marker that - task failures also use, so it can be stored in ``completed_steps``.""" + task failures also use, so it can be stored in ``completed_steps``. + + The marker's final shape is decided by the backend (``wac_failure_record``), + which normalizes task failures through the same function; what is built here + is the raw material plus the envelope the backend recognizes.""" + error = {"name": type(exc).__name__, "message": _safe_str(exc)} + stack = _step_error_stack(exc) + if stack: + error["stack"] = stack + # Custom attributes go under ``extra``, the same key the python executor uses + # for a failed child job, so an exception carrying e.g. a ``code`` keeps it + # whether it failed as a task or as a step. + # + # Coerced through ``default=str`` the way the executor writes its own error: + # the fast-path POST serializes strictly, and the commonest failing step + # there is — ``resp.raise_for_status()``, whose ``__dict__`` holds a request + # and a response object — would otherwise fail to serialize and silently + # drop every such failure onto the slow suspend-and-replay path. + # ``getattr``'s default only swallows ``AttributeError``; an exception + # overriding ``__getattribute__`` raises whatever it likes from this read. + try: + extra = getattr(exc, "__dict__", None) + except Exception: + extra = None + if extra: + # ``default`` is where json hands back the objects it cannot represent, + # and ``str()`` on a detached ORM row or a proxy over a closed + # connection raises in turn. This runs inside the ``except`` that is + # reporting the user's failure, so an escape here would replace their + # error with an unrelated one and skip the checkpoint entirely. + # Narrow: what json raises for something it cannot represent. A broader + # catch would hide a mistake in this function as a silently missing + # field, which is how it read before. + try: + # ``parse_constant`` catches the one thing ``default`` cannot: a + # float is serializable, so ``NaN``/``Infinity`` pass through as + # bare literals that are not JSON and that the backend's extractor + # rejects — taking the whole checkpoint down with them. Kept as + # their text rather than dropped, so the attribute still says + # something. + error["extra"] = json.loads( + json.dumps(extra, default=_safe_str), parse_constant=lambda c: c + ) + except (TypeError, ValueError, RecursionError): + pass return { "__wmill_error": True, - "message": str(exc), + "message": _safe_str(exc), "step_key": key, - "result": {"error": str(exc), "type": type(exc).__name__}, + "result": {"error": error}, } -def _step_error_from_marker(marker: dict, name: str) -> TaskError: - """Rebuild the exception a failed step raises. Both the run that produced the - failure and every later replay go through here, so a workflow's ``except`` - clauses see the same type either way.""" +def _task_error_from_marker(marker: dict, fallback_message: str) -> TaskError: + """Rebuild the exception a failed task or step raises. The run that produced + the failure and every later replay go through here: ``except`` is control + flow, ``@workflow`` re-runs its body from the top every round, so a handler + that branches on the failure it caught must be handed the same thing in + every round or it dispatches different tasks on the way back.""" return TaskError( - marker.get("message", f"Step '{name}' failed"), + marker.get("message") or fallback_message, step_key=marker.get("step_key", ""), - child_job_id=marker.get("child_job_id", ""), + child_job_id=marker.get("child_job_id"), result=marker.get("result"), ) @@ -2792,12 +2881,7 @@ class WorkflowCtx: if key in self._completed: val = self._completed[key] if isinstance(val, dict) and val.get("__wmill_error"): - raise TaskError( - val.get("message", f"Task '{name}' failed"), - step_key=val.get("step_key", ""), - child_job_id=val.get("child_job_id", ""), - result=val.get("result"), - ) + raise _task_error_from_marker(val, f"Task '{name}' failed") return self._resolved(val) if self._executing_key is not None: @@ -2904,7 +2988,7 @@ class WorkflowCtx: if key in self._completed: val = self._completed[key] if isinstance(val, dict) and val.get("__wmill_error"): - raise _step_error_from_marker(val, name) + raise _task_error_from_marker(val, f"Step '{name}' failed") return val if self._executing_key is not None: @@ -2916,16 +3000,25 @@ class WorkflowCtx: t0 = _time_mod.monotonic() # A raised step still has to reach ``completed_steps``, or a replay with # ``_executing_key`` set finds nothing recorded and parks forever on the - # ``_asyncio.Future()`` above. ``_StepSuspend`` and ``CancelledError`` are - # ``BaseException``, so they pass through untouched. - step_error: Optional[Exception] = None + # ``_asyncio.Future()`` above. The control-flow signals (``_StepSuspend``, + # ``_StepFailure``) and ``CancelledError`` are ``BaseException``, so they + # pass through untouched. + step_failed = False try: result = fn() if _asyncio.iscoroutine(result): result = await result except Exception as _exc: - step_error = _exc + step_failed = True result = _step_error_marker(key, _exc) + # The failure is reported as a value from here on, so nothing else + # prints the traceback. Without this a step that fails and is never + # caught leaves a job log whose deepest frame is inside this client. + print(f"--- WAC: {key} failed ---") + print(f"{type(_exc).__name__}: {_safe_str(_exc)}") + _step_stack = result["result"]["error"].get("stack") + if _step_stack: + print(_step_stack) duration_ms = int((_time_mod.monotonic() - t0) * 1000) # Fast path: POST the delta to the new per-job API endpoint and return @@ -2943,6 +3036,7 @@ class WorkflowCtx: _token = os.environ.get("WM_TOKEN") if _fast_path_enabled and _job_id and _workspace and _base and _token: _fast_path_ok = False + _stored_failure = None _replay_result = None try: # ``default=str`` is the encoder the worker wrapper uses on the @@ -2978,6 +3072,20 @@ class WorkflowCtx: content=_payload, ) _resp.raise_for_status() + if step_failed: + # The backend normalizes the failure before storing it, + # and hands back what it stored. Raising from that, not + # from the marker posted above, is what makes this round + # and every replay read the same record even if the two + # sides ever disagree about how to build one. + # + # A backend predating the echo answers without a JSON + # body, and the round-tripped marker below stands in. A + # JSON body that will not parse is different: the record + # may already be committed and its content is unknown, + # so let it raise and take the suspend path instead. + if "json" in _resp.headers.get("content-type", ""): + _stored_failure = (_resp.json() or {}).get("failure") _fast_path_ok = True except Exception as _e: logger.info( @@ -2987,12 +3095,21 @@ class WorkflowCtx: ) # fall through to the legacy suspend path if _fast_path_ok: - # Raise what a replay would rebuild from the marker, never the + # Raise what a replay would rebuild from the record, never the # original: a replay cannot reconstruct the original type, so # raising it here would make ``except ValueError:`` catch on this - # run and miss on the next. ``__cause__`` is for tracebacks only. - if step_error is not None: - raise _step_error_from_marker(result, name) from step_error + # run and miss on the next. Nothing is chained onto + # ``__cause__`` for the same reason — the traceback a replay can + # still show is in ``result["error"]["stack"]``. ``_stored_failure`` + # is None against a backend that predates the echoed record, + # which is what ``_replay_result`` below stands in for. + if step_failed: + # ``_replay_result``, not ``result``: the fallback has to be + # what the checkpoint holds, so the round that ran the body + # reads what every replay of it will. + raise _task_error_from_marker( + _stored_failure or _replay_result, f"Step '{name}' failed" + ) # Return the round trip of what was checkpointed, never the # in-memory value: handing back the live object would let the # round that ran the body branch on a type — tuple, datetime — diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 3d0124b2a8..dec67ba4cb 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -1073,6 +1073,8 @@ Let task errors fail the workflow unless the user asks for recovery logic. Python: \`except Exception\` is safe around WAC calls because internal suspension inherits from \`BaseException\`. Avoid bare \`except:\` in workflow code. If the user asks for recovery logic around failed child work, catch \`TaskError\` from \`wmill\` for task failures. TypeScript: avoid broad \`try/catch\` around WAC SDK calls. The SDK uses an internal suspension error during initial dispatch; catching it can break workflow suspension. If a broad catch is unavoidable, rethrow internal suspension errors before handling business errors. + +A caught failure reads the same whether it came from a task or from a \`step()\`, and the same in the round that ran the failing body as in every round replaying it. It carries \`step_key\`, \`child_job_id\` (absent for a \`step()\`, which runs in the workflow job and has no child job), a \`message\` that is the failure's own message, and \`result\` = \`{"error": {"name", "message", "stack"?, "extra"?}}\`. \`name\`, \`message\` and \`stack\` are the fields that read the same whichever side failed; \`name\` and \`message\` are always there, \`stack\` only when the failure had a traceback to give. \`extra\` carries the failure's own custom fields (an exception's attributes, an error's properties) and is best-effort: it is absent when there were none, and a task can report entries a step does not, so read it defensively and don't branch on its absence. \`extra\` is dropped when it is too large to keep in the checkpoint, and \`extra_omitted: true\` says so — absent \`extra\` with no \`extra_omitted\` means the failure simply had no custom fields. Branch on those, not on the original exception type: the workflow body re-runs from the top every round and a replay rebuilds the failure from the checkpoint, so nothing outside that record survives. Python raises \`TaskError\`; TypeScript throws an \`Error\` named \`TaskError\` carrying the same fields. Nothing is chained onto \`__cause__\` / \`cause\` — the traceback is in \`result.error.stack\`, and is also printed to the job log when the step fails. `; export const FLOW_CHAT_SPECIAL_MODULES = `## Special Modules @@ -2622,14 +2624,19 @@ export const WAC_SDK_PYTHON = `## Python Workflow-as-Code API (wmill) Import: \`from wmill import workflow, task, task_script, task_flow, step, sleep, wait_for_approval, get_approval_urls, get_resume_urls, parallel, TaskError\` \`\`\`python -# Raised when a WAC task step failed. +# Raised when a WAC \`\`task\`\` or \`\`step\`\` failed. # # Attributes: # step_key: The checkpoint key of the failed step. -# child_job_id: The UUID of the failed child job. -# result: The error result from the child job. +# child_job_id: The UUID of the failed child job, or \`\`None\`\` for a +# \`\`step()\`\`, which runs in the workflow job and has no child job. +# result: \`\`{"error": {"name", "message", "stack"?, "extra"?}}\`\` — the +# same shape whether a task or a step failed. \`\`name\`\` and \`\`message\`\` +# are always present; \`\`stack\`\` only when the failure had a traceback, +# and \`\`extra\`\` only when it carried custom fields of its own, dropped +# with \`\`extra_omitted: True\`\` beside it when too large to checkpoint. class TaskError(Exception): - def __init__(self, message: str, *, step_key: str = '', child_job_id: str = '', result = None) + def __init__(self, message: str, *, step_key: str = '', child_job_id: Optional[str] = None, result = None) # Get URLs needed for resuming a flow after suspension. # diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index 6d0c34deb4..e1ac8635f8 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -3,14 +3,19 @@ Import: `from wmill import workflow, task, task_script, task_flow, step, sleep, wait_for_approval, get_approval_urls, get_resume_urls, parallel, TaskError` ```python -# Raised when a WAC task step failed. +# Raised when a WAC ``task`` or ``step`` failed. # # Attributes: # step_key: The checkpoint key of the failed step. -# child_job_id: The UUID of the failed child job. -# result: The error result from the child job. +# child_job_id: The UUID of the failed child job, or ``None`` for a +# ``step()``, which runs in the workflow job and has no child job. +# result: ``{"error": {"name", "message", "stack"?, "extra"?}}`` — the +# same shape whether a task or a step failed. ``name`` and ``message`` +# are always present; ``stack`` only when the failure had a traceback, +# and ``extra`` only when it carried custom fields of its own, dropped +# with ``extra_omitted: True`` beside it when too large to checkpoint. class TaskError(Exception): - def __init__(self, message: str, *, step_key: str = '', child_job_id: str = '', result = None) + def __init__(self, message: str, *, step_key: str = '', child_job_id: Optional[str] = None, result = None) # Get URLs needed for resuming a flow after suspension. # diff --git a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md index 159b3636e3..d02b0a90cf 100644 --- a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md +++ b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md @@ -238,6 +238,8 @@ Python: `except Exception` is safe around WAC calls because internal suspension TypeScript: avoid broad `try/catch` around WAC SDK calls. The SDK uses an internal suspension error during initial dispatch; catching it can break workflow suspension. If a broad catch is unavoidable, rethrow internal suspension errors before handling business errors. +A caught failure reads the same whether it came from a task or from a `step()`, and the same in the round that ran the failing body as in every round replaying it. It carries `step_key`, `child_job_id` (absent for a `step()`, which runs in the workflow job and has no child job), a `message` that is the failure's own message, and `result` = `{"error": {"name", "message", "stack"?, "extra"?}}`. `name`, `message` and `stack` are the fields that read the same whichever side failed; `name` and `message` are always there, `stack` only when the failure had a traceback to give. `extra` carries the failure's own custom fields (an exception's attributes, an error's properties) and is best-effort: it is absent when there were none, and a task can report entries a step does not, so read it defensively and don't branch on its absence. `extra` is dropped when it is too large to keep in the checkpoint, and `extra_omitted: true` says so — absent `extra` with no `extra_omitted` means the failure simply had no custom fields. Branch on those, not on the original exception type: the workflow body re-runs from the top every round and a replay rebuilds the failure from the checkpoint, so nothing outside that record survives. Python raises `TaskError`; TypeScript throws an `Error` named `TaskError` carrying the same fields. Nothing is chained onto `__cause__` / `cause` — the traceback is in `result.error.stack`, and is also printed to the job log when the step fails. + ## TypeScript Workflow-as-Code API (windmill-client) @@ -376,14 +378,19 @@ export async function parallel(items: T[], fn: (item: T) => PromiseLike Import: `from wmill import workflow, task, task_script, task_flow, step, sleep, wait_for_approval, get_approval_urls, get_resume_urls, parallel, TaskError` ```python -# Raised when a WAC task step failed. +# Raised when a WAC ``task`` or ``step`` failed. # # Attributes: # step_key: The checkpoint key of the failed step. -# child_job_id: The UUID of the failed child job. -# result: The error result from the child job. +# child_job_id: The UUID of the failed child job, or ``None`` for a +# ``step()``, which runs in the workflow job and has no child job. +# result: ``{"error": {"name", "message", "stack"?, "extra"?}}`` — the +# same shape whether a task or a step failed. ``name`` and ``message`` +# are always present; ``stack`` only when the failure had a traceback, +# and ``extra`` only when it carried custom fields of its own, dropped +# with ``extra_omitted: True`` beside it when too large to checkpoint. class TaskError(Exception): - def __init__(self, message: str, *, step_key: str = '', child_job_id: str = '', result = None) + def __init__(self, message: str, *, step_key: str = '', child_job_id: Optional[str] = None, result = None) # Get URLs needed for resuming a flow after suspension. # diff --git a/system_prompts/base/workflow-as-code.md b/system_prompts/base/workflow-as-code.md index f0b72ead96..9a1eea4700 100644 --- a/system_prompts/base/workflow-as-code.md +++ b/system_prompts/base/workflow-as-code.md @@ -185,3 +185,5 @@ Let task errors fail the workflow unless the user asks for recovery logic. Python: `except Exception` is safe around WAC calls because internal suspension inherits from `BaseException`. Avoid bare `except:` in workflow code. If the user asks for recovery logic around failed child work, catch `TaskError` from `wmill` for task failures. TypeScript: avoid broad `try/catch` around WAC SDK calls. The SDK uses an internal suspension error during initial dispatch; catching it can break workflow suspension. If a broad catch is unavoidable, rethrow internal suspension errors before handling business errors. + +A caught failure reads the same whether it came from a task or from a `step()`, and the same in the round that ran the failing body as in every round replaying it. It carries `step_key`, `child_job_id` (absent for a `step()`, which runs in the workflow job and has no child job), a `message` that is the failure's own message, and `result` = `{"error": {"name", "message", "stack"?, "extra"?}}`. `name`, `message` and `stack` are the fields that read the same whichever side failed; `name` and `message` are always there, `stack` only when the failure had a traceback to give. `extra` carries the failure's own custom fields (an exception's attributes, an error's properties) and is best-effort: it is absent when there were none, and a task can report entries a step does not, so read it defensively and don't branch on its absence. `extra` is dropped when it is too large to keep in the checkpoint, and `extra_omitted: true` says so — absent `extra` with no `extra_omitted` means the failure simply had no custom fields. Branch on those, not on the original exception type: the workflow body re-runs from the top every round and a replay rebuilds the failure from the checkpoint, so nothing outside that record survives. Python raises `TaskError`; TypeScript throws an `Error` named `TaskError` carrying the same fields. Nothing is chained onto `__cause__` / `cause` — the traceback is in `result.error.stack`, and is also printed to the job log when the step fails. diff --git a/system_prompts/generate.py b/system_prompts/generate.py index 0c11a86b1b..74e0cf35a2 100644 --- a/system_prompts/generate.py +++ b/system_prompts/generate.py @@ -107,6 +107,12 @@ def extract_ts_functions(content: str) -> list[dict]: if not return_type: return_type = 'Promise' if is_async else 'void' + # `@internal` marks an export that exists for another module or for a + # test to reach, not for a user to call. The SDK reference these prompts + # become is a user-facing API list, so it must not advertise them. + if jsdoc_raw and '@internal' in jsdoc_raw: + continue + docstring = clean_jsdoc(jsdoc_raw) if jsdoc_raw else '' seen_names.add(name) functions.append({ diff --git a/typescript-client/build.jsr.sh b/typescript-client/build.jsr.sh index a8e3cdc390..821c4f8a0c 100755 --- a/typescript-client/build.jsr.sh +++ b/typescript-client/build.jsr.sh @@ -10,6 +10,7 @@ npx --yes @hey-api/openapi-ts@0.43.0 --input "${script_dirpath}/../backend/wind sed -i 's/get \[Symbol\.toStringTag\]() {/get \[Symbol\.toStringTag\]() : string {/g' "${script_dirpath}/src/core/CancelablePromise.ts" cp "${script_dirpath}/client.ts" "${script_dirpath}/src/" +cp "${script_dirpath}/wacError.ts" "${script_dirpath}/src/" cp "${script_dirpath}/s3Types.ts" "${script_dirpath}/src/" cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" diff --git a/typescript-client/build.sh b/typescript-client/build.sh index 7e1c52335c..db45f81b36 100755 --- a/typescript-client/build.sh +++ b/typescript-client/build.sh @@ -35,6 +35,7 @@ fi cp "${script_dirpath}/client.ts" "${script_dirpath}/src/" +cp "${script_dirpath}/wacError.ts" "${script_dirpath}/src/" cp "${script_dirpath}/s3Types.ts" "${script_dirpath}/src/" cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 4a31d2dfed..3c7159fb94 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -12,6 +12,7 @@ import { KafkaTriggerService, } from "./services.gen"; import { OpenAPI } from "./core/OpenAPI"; +import { isSuspendSignal, stepErrorMarker, taskErrorFromMarker } from "./wacError"; // import type { DenoS3LightClientSettings } from "./index"; import { DenoS3LightClientSettings, @@ -1520,32 +1521,6 @@ export class StepSuspend extends Error { } } -/** Serialize a failed `step()` body into the `__wmill_error` marker that task - * failures also use, so it can be stored in `completed_steps`. */ -function stepErrorMarker(key: string, e: unknown): Record { - const message = e instanceof Error ? e.message : String(e); - // Constructor name, not `e.name`: a `class MyError extends Error {}` that - // never assigns `this.name` reports "Error", which would make the same - // failure read as `MyError` in the python client and `Error` here. - const type = e instanceof Error ? (e.constructor?.name ?? e.name) : typeof e; - return { __wmill_error: true, message, step_key: key, result: { error: message, type } }; -} - -/** Rebuild the error a failed step throws. Both the run that produced the - * failure and every later replay go through here, so a workflow's catch block - * sees the same shape either way. */ -function stepErrorFromMarker(marker: any, name: string): Error { - const err = new Error(marker?.message || `Step '${name}' failed`); - // Matches the python client, which raises TaskError here; the failed body's - // own type stays in `result.type`. Keeps a failed job's serialized error - // identical across the two languages. - err.name = "TaskError"; - (err as any).result = marker?.result; - (err as any).step_key = marker?.step_key; - (err as any).child_job_id = marker?.child_job_id; - return err; -} - /** Values `JSON.stringify` cannot represent: it omits the property holding one. * The type-level half of `checkpointableResult`, which nulls them at the top * level, where there is no key to omit. */ @@ -1763,11 +1738,7 @@ export class WorkflowCtx { if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { - const err = new Error((value as any).message || `Task '${name}' failed`); - err.name = "TaskError"; - (err as any).result = (value as any).result; - (err as any).step_key = (value as any).step_key; - (err as any).child_job_id = (value as any).child_job_id; + const err = taskErrorFromMarker(value, `Task '${name}' failed`); return { then: (_resolve: any, reject?: any) => { if (reject) reject(err); else throw err; } } as PromiseLike; } return { then: (resolve: any) => resolve(value) }; @@ -1894,7 +1865,7 @@ export class WorkflowCtx { if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { - throw stepErrorFromMarker(value, name); + throw taskErrorFromMarker(value, `Step '${name}' failed`); } return value as T; } @@ -1912,15 +1883,24 @@ export class WorkflowCtx { // never-resolving promise above. A nested StepSuspend is control flow, // not a step failure. let result: T; - let stepError: unknown; let errored = false; try { result = await fn(); } catch (e) { - if ((e as any)?.name === "StepSuspend" || e instanceof StepSuspend) throw e; + if (isSuspendSignal(e, StepSuspend)) throw e; errored = true; - stepError = e; result = stepErrorMarker(key, e) as any; + // The failure is reported as a value from here on, so nothing else prints + // the stack. Without this a step that throws and is never caught leaves a + // job log whose deepest frame is inside this client. + console.log(`--- WAC: ${key} failed ---`); + // Only from the marker, which already coerced the thrown value once and + // survived it. Reaching for `e` again here would re-run the coercion that + // `stepErrorMarker` guards, and throw out of this catch before the failure + // is ever checkpointed. + console.log( + (result as any)?.result?.error?.stack ?? (result as any)?.message ?? "step failed", + ); } const durationMs = Date.now() - t0; @@ -1988,6 +1968,22 @@ export class WorkflowCtx { if (!resp.ok) { throw new Error(`inline_checkpoint API ${resp.status}`); } + if (!errored) return undefined; + // The backend normalizes the failure before storing it, and hands + // back what it stored. Throwing from that, not from the marker posted + // above, is what makes this round and every replay read the same + // record even if the two sides ever disagree about how to build one. + // + // A backend predating the echo answers without a JSON body; the + // caller then falls back to the round trip of what it posted. A JSON + // body we cannot read is different: the normalized record may already + // be committed and we do not know what it says, so let this reject + // and take the suspend path, where the next round reads whatever the + // backend actually stored. + if (!(resp.headers.get("content-type") ?? "").includes("json")) { + return undefined; + } + return (await resp.json())?.failure; } finally { clearTimeout(t); } @@ -1995,8 +1991,9 @@ export class WorkflowCtx { // Swallow chain errors so a past failure does not poison future awaits. this._inlineChain = chainTail.catch(() => {}); let fastPathOk = false; + let storedFailure: any; try { - await chainTail; + storedFailure = await chainTail; fastPathOk = true; } catch (e) { console.log( @@ -2005,12 +2002,17 @@ export class WorkflowCtx { // fall through to the legacy suspend path below } if (fastPathOk) { - // Throw what a replay would rebuild from the marker, never the + // Throw what a replay would rebuild from the record, never the // original: a replay cannot reconstruct the original type, so throwing // it here would match `e instanceof TypeError` on this run and miss on - // the next. `cause` is for logging only — absent on replay. + // the next. Nothing is attached to `cause` for the same reason — the + // stack a replay can still show is in `result.error.stack`. if (errored) { - throw Object.assign(stepErrorFromMarker(result, name), { cause: stepError }); + // `checkpointed`, not `result`: against a backend with no echo the + // fallback still has to be what the checkpoint holds, or an `extra` + // carrying a Date reads as a Date now and as its ISO string on every + // replay — the divergence the success path below already avoids. + throw taskErrorFromMarker(storedFailure ?? checkpointed, `Step '${name}' failed`); } // Return the round trip of what was checkpointed, never the in-memory // value: handing back the live object would let the round that ran the @@ -2155,7 +2157,7 @@ export function task Promise>( try { result = await fn(...args); } catch (e) { - if ((e as any)?.name === "StepSuspend" || e instanceof StepSuspend) throw e; + if (isSuspendSignal(e, StepSuspend)) throw e; ctx._raiseStepFailure(e); } ctx._raiseSuspend({ mode: "step_complete", steps: [], result: checkpointableResult(result) }); diff --git a/typescript-client/jsr.json b/typescript-client/jsr.json index 4c58c5779a..2b0d66fa25 100644 --- a/typescript-client/jsr.json +++ b/typescript-client/jsr.json @@ -3,7 +3,7 @@ "version": "1.773.0", "exports": "./src/index.ts", "publish": { - "exclude": ["!src", "./s3Types.ts", "./sqlUtils.ts", "./client.ts"] + "exclude": ["!src", "./s3Types.ts", "./sqlUtils.ts", "./client.ts", "./wacError.ts"] }, "license": "Apache-2.0" } diff --git a/typescript-client/tests/workflow.test.ts b/typescript-client/tests/workflow.test.ts index 48a20b5374..179244314b 100644 --- a/typescript-client/tests/workflow.test.ts +++ b/typescript-client/tests/workflow.test.ts @@ -5,6 +5,11 @@ */ import { expect, test, describe } from "bun:test"; +// The two functions that decide what a caught failure looks like come from the +// shipped module, not from the mirror below: they are what drifted from the +// backend's shape before, so a copy of them here would guard nothing. +import { isSuspendSignal, stepErrorMarker, taskErrorFromMarker } from "../wacError"; + // --- Inline SDK (mirrors client.ts implementation) --- class StepSuspend extends Error { @@ -79,11 +84,7 @@ class WorkflowCtx { if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { - const err = new Error((value as any).message || `Task '${name}' failed`); - err.name = "TaskError"; - (err as any).result = (value as any).result; - (err as any).step_key = (value as any).step_key; - (err as any).child_job_id = (value as any).child_job_id; + const err = taskErrorFromMarker(value, `Task '${name}' failed`); return { then: (_resolve: any, reject?: any) => { if (reject) reject(err); else throw err; } }; } return { then: (resolve: any) => resolve(value) }; @@ -157,10 +158,7 @@ class WorkflowCtx { if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { - const err = new Error((value as any).message || `Step '${name}' failed`); - err.name = "TaskError"; - (err as any).result = (value as any).result; - throw err; + throw taskErrorFromMarker(value, `Step '${name}' failed`); } return value as T; } @@ -170,22 +168,11 @@ class WorkflowCtx { } let result: any; - let errored = false; try { result = await fn(); } catch (e: any) { - if (e?.name === "StepSuspend" || e instanceof StepSuspend) throw e; - errored = true; - const message = e instanceof Error ? e.message : String(e); - result = { - __wmill_error: true, - message, - step_key: key, - result: { - error: message, - type: e instanceof Error ? (e.constructor?.name ?? e.name) : typeof e, - }, - }; + if (isSuspendSignal(e, StepSuspend)) throw e; + result = stepErrorMarker(key, e); } this._raiseSuspend({ mode: "inline_checkpoint", @@ -248,7 +235,7 @@ function task Promise>( try { result = await fn(...args); } catch (e: any) { - if (e?.name === "StepSuspend" || e instanceof StepSuspend) throw e; + if (isSuspendSignal(e, StepSuspend)) throw e; ctx._raiseStepFailure(e); } ctx._raiseSuspend({ @@ -1415,11 +1402,18 @@ describe("error propagation via __wmill_error marker", () => { // _executingKey set, reaches the unrecorded key, and parks on the // never-resolving promise forever. describe("throwing inline step is checkpointed", () => { + // A failed child job reports `{ error: { name, message, stack } }`, and a + // failed step() has to be indistinguishable from it. The stack is a + // JS stack string, asserted separately. const marker = { __wmill_error: true, message: "boom", step_key: "step_0", - result: { error: "boom", type: "TypeError" }, + result: { error: { name: "TypeError", message: "boom" } }, + }; + const withoutStack = (m: any) => { + const { stack, ...error } = m.result.error; + return { ...m, result: { ...m.result, error } }; }; // The workflow body catches — the shape a failing step is written for, and @@ -1450,7 +1444,8 @@ describe("throwing inline step is checkpointed", () => { expect(caught).toBeInstanceOf(StepSuspend); expect(caught.dispatchInfo.mode).toBe("inline_checkpoint"); expect(caught.dispatchInfo.key).toBe("step_0"); - expect(caught.dispatchInfo.result).toEqual(marker); + expect(withoutStack(caught.dispatchInfo.result)).toEqual(marker); + expect(caught.dispatchInfo.result.result.error.stack).toContain("TypeError: boom"); }); test("a swallowed suspend still reaches the runner", async () => { @@ -1459,7 +1454,7 @@ describe("throwing inline step is checkpointed", () => { const result = await runWorkflow(catchingWf(), {}, [5]); expect(result.type).toBe("inline_checkpoint"); expect(result.key).toBe("step_0"); - expect(result.result).toEqual(marker); + expect(withoutStack(result.result)).toEqual(marker); }); test("a swallowed suspend from a succeeding step still reaches the runner", async () => { @@ -1476,6 +1471,176 @@ describe("throwing inline step is checkpointed", () => { expect(result.result).toBe(42); }); + // The backend records `name: e.name` for a failed child job + // (bun_executor.rs). A step reporting the constructor name instead would make + // the same failure read differently depending on whether it ran as a task or + // as a step, in the one field handlers are told to branch on. + test("error.name is e.name, as a failed child job reports it", async () => { + class MyError extends Error {} + const unnamed = stepErrorMarker("k", new MyError("boom")); + expect(unnamed.result.error.name).toBe("Error"); + + class NamedError extends Error { + name = "NamedError"; + } + const named = stepErrorMarker("k", new NamedError("boom")); + expect(named.result.error.name).toBe("NamedError"); + + const domish = Object.assign(new Error("aborted"), { name: "AbortError" }); + expect(stepErrorMarker("k", domish).result.error.name).toBe("AbortError"); + }); + + // A failed child job reports custom properties under `error.extra` + // (bun_executor.rs). A step dropping them would make the same error carry + // less information depending on how it was run. + test("custom error properties survive under error.extra, as a task's do", async () => { + const e = Object.assign(new Error("429"), { code: 429, retryAfter: 5 }); + const marker = stepErrorMarker("k", e); + expect(marker.result.error.extra).toEqual({ code: 429, retryAfter: 5 }); + // the named fields the executors report separately are not duplicated + expect(marker.result.error.extra.message).toBeUndefined(); + expect(marker.result.error.extra.stack).toBeUndefined(); + + expect(stepErrorMarker("k", new Error("plain")).result.error.extra).toBeUndefined(); + }); + + // The marker is stringified while the workflow is still running (checkpoint + // POST, then wrapper output). A property that can't survive that would end + // the job instead of reaching the user's catch. + test("a property that cannot be serialized is dropped, not propagated", () => { + const circular: any = { name: "req" }; + circular.self = circular; + const withCircular = Object.assign(new Error("boom"), { code: 429, request: circular }); + const marker = stepErrorMarker("k", withCircular); + expect(() => JSON.stringify(marker)).not.toThrow(); + expect(marker.result.error.extra).toEqual({ code: 429 }); + + const withThrowingAccessor = new Error("boom"); + Object.defineProperty(withThrowingAccessor, "boobytrap", { + enumerable: true, + get() { + throw new Error("read me and die"); + }, + }); + expect(() => stepErrorMarker("k", withThrowingAccessor)).not.toThrow(); + }); + + // A task executor reads `name`/`message`/`stack` off whatever was thrown, not + // off an Error instance, so a step must too or a handler can tell the two + // apart in the fields the contract tells it to branch on. + test("a non-Error throw records what a task would record", () => { + const thrown = stepErrorMarker("k", { name: "Thrown", message: "boom", code: 429 }); + expect(thrown.result.error.name).toBe("Thrown"); + expect(thrown.result.error.message).toBe("boom"); + expect(thrown.result.error.extra).toEqual({ code: 429 }); + + // A string carries none of the three, so the record is left for the backend + // to fill — the same fallback a task throwing a string produces. Its + // character indices are not custom fields. + const str = stepErrorMarker("k", "boom"); + expect(str.result.error).toEqual({}); + + // `String()` on a value with no `toString` to reach throws in turn, and + // this runs inside the catch reporting the user's failure. + expect(() => stepErrorMarker("k", Object.create(null))).not.toThrow(); + }); + + // Reporting a failure must not be able to fail: every read of the thrown + // value happens inside the catch that is reporting it, so an escape replaces + // the user's error with an unrelated one and leaves the step uncheckpointed. + test("a hostile thrown value cannot make failure reporting throw", () => { + const hostile = new Proxy( + {}, + { + get() { + throw new Error("get trap"); + }, + ownKeys() { + throw new Error("ownKeys trap"); + }, + }, + ); + expect(() => stepErrorMarker("k", hostile)).not.toThrow(); + expect(() => JSON.stringify(stepErrorMarker("k", hostile))).not.toThrow(); + + const throwingToString = { toString() { throw new Error("no"); } }; + expect(() => stepErrorMarker("k", throwingToString)).not.toThrow(); + }); + + // Probing the original value only proves it serialized once. The marker is + // serialized again to reach the checkpoint, and by then the failure has + // nowhere left to go, so what survived the probe is what gets kept. + test("a property that serializes only once cannot break the checkpoint", () => { + let calls = 0; + const onceOnly = { + toJSON() { + if (calls++ > 0) throw new Error("second time"); + return { ok: true }; + }, + }; + const marker = stepErrorMarker("k", Object.assign(new Error("boom"), { payload: onceOnly })); + expect(marker.result.error.extra.payload).toEqual({ ok: true }); + // re-serialized on the way to the checkpoint, and again by the wrapper + expect(() => JSON.stringify(marker)).not.toThrow(); + expect(() => JSON.stringify(marker)).not.toThrow(); + }); + + // The caller reads `.name` off the thrown value to spot a suspend, before it + // ever reaches the hardened marker. A hostile value escaping there leaves the + // step uncheckpointed and a later replay parks on it forever. + test("a hostile throw still reaches the checkpoint through _runInlineStep", async () => { + const hostile = new Proxy( + {}, + { + get() { + throw new Error("get trap"); + }, + ownKeys() { + throw new Error("ownKeys trap"); + }, + }, + ); + const ctx = new WorkflowCtx({}); + let caught: any; + try { + await ctx._runInlineStep("risky", () => { + throw hostile; + }); + } catch (e) { + caught = e; + } + // the suspend carrying the checkpoint, not the hostile value itself + expect(caught).toBeInstanceOf(StepSuspend); + expect(caught.dispatchInfo.key).toBe("step_0"); + expect(caught.dispatchInfo.result.__wmill_error).toBe(true); + }); + + // `instanceof` consults a proxy's `getPrototypeOf` trap, so the suspend check + // itself can throw — before anything is checkpointed. + test("suspend detection survives a value that refuses to be inspected", () => { + const hostilePrototype = new Proxy(new StepSuspend({ mode: "sequential" }), { + getPrototypeOf() { + throw new Error("getPrototypeOf trap"); + }, + }); + // the name is still readable, so it is still recognised as the signal + expect(isSuspendSignal(hostilePrototype, StepSuspend)).toBe(true); + + const opaque = new Proxy( + {}, + { + getPrototypeOf() { + throw new Error("getPrototypeOf trap"); + }, + get() { + throw new Error("get trap"); + }, + }, + ); + expect(() => isSuspendSignal(opaque, StepSuspend)).not.toThrow(); + expect(isSuspendSignal(opaque, StepSuspend)).toBe(false); + }); + test("a replayed step failure is named TaskError, like the python client", async () => { const ctx = new WorkflowCtx({ completed_steps: { step_0: marker } }); let caught: any; @@ -1485,8 +1650,15 @@ describe("throwing inline step is checkpointed", () => { caught = e; } expect(`${caught.name}: ${caught.message}`).toBe("TaskError: boom"); - // the failing body's own type stays addressable here - expect(caught.result).toEqual({ error: "boom", type: "TypeError" }); + // the failing body's own type stays addressable here, in the shape a + // failed task hands over too + expect(caught.result).toEqual({ error: { name: "TypeError", message: "boom" } }); + expect(caught.step_key).toBe("step_0"); + // a step runs in the workflow job, so there is no child job to name + expect(caught.child_job_id).toBeUndefined(); + // nothing is chained onto `cause`: a replay has no original error to + // chain, so neither round does + expect(caught.cause).toBeUndefined(); }); test("a child job cannot swallow its own completion signal", async () => { diff --git a/typescript-client/wacError.ts b/typescript-client/wacError.ts new file mode 100644 index 0000000000..025de78a7b --- /dev/null +++ b/typescript-client/wacError.ts @@ -0,0 +1,159 @@ +// Workflow-as-Code failure record, client side. +// +// Deliberately dependency-free so the tests can import it directly: the rest of +// client.ts pulls in the generated API modules, which is why the workflow test +// suite re-implements WorkflowCtx inline. What a caught failure looks like, and +// what is a failure at all rather than the SDK's own control flow, is decided +// here — so it is decided against the shipped code rather than against a copy. + +/** Error properties the executors already report as named fields, so they must + * not be repeated inside `extra`. Kept identical to the skip-list in the + * generated bun/deno job wrappers. */ +const SERIALIZED_ERROR_FIELDS = [ + "line", + "name", + "stack", + "column", + "message", + "sourceURL", + "originalLine", + "originalColumn", +]; + +/** Read a property off a value that may fight back: a proxy `get` trap or a + * throwing accessor turns an ordinary field read into an exception. Callers + * are on the failure-reporting path, where nothing has been checkpointed yet. */ +function safeRead(o: any, k: string): unknown { + try { + return o?.[k]; + } catch { + return undefined; + } +} + +/** @internal + * Whether a caught value is the SDK's own suspend signal, which must be + * rethrown rather than reported as a step failure. + * + * Both halves of the check can throw on a hostile value: `instanceof` consults + * a proxy's `getPrototypeOf` trap, and reading `.name` its `get` trap. This + * runs before anything has been checkpointed, so neither may escape — a value + * that fights back is simply not a suspend. */ +export function isSuspendSignal(e: unknown, suspendCtor: Function): boolean { + try { + if (e instanceof (suspendCtor as any)) return true; + } catch { + // a proxy refusing to be inspected is not the SDK's own signal + } + return safeRead(e, "name") === "StepSuspend"; +} + +/** @internal + * Serialize a failed `step()` body into the `__wmill_error` marker that task + * failures also use, so it can be stored in `completed_steps`. + * + * The record's final shape is decided by the backend (`wac_failure_record`), + * which normalizes task failures through the same function; what is built here + * is the raw material plus the envelope the backend recognizes. */ +export function stepErrorMarker(key: string, e: unknown): Record { + const thrown = e as any; + // The three named fields are read off whatever was thrown, exactly as the + // bun/deno wrappers read them for a failed child job — `e.name`, `e.message`, + // `e.stack`, whether or not it is an `Error`. Anything else here lets a + // handler tell a task from a step in the fields it is told to branch on: a + // thrown `{ name, message, code }` keeps all three as a task, and a + // `class MyError extends Error {}` that never assigns `this.name` reports + // `Error` as a task, so both must here too. What is absent stays absent and + // the backend fills it, which is what a task with no usable fields gets. + // + // `throw "boom"` therefore reports the backend's fallback message rather than + // "boom", because a task throwing a string is equally lossy — its `e.message` + // is undefined too. Recovering the text here, from the marker's top-level + // `message`, would split `message` between a task and a step, which is the + // thing this record exists to prevent. The executors are where a primitive + // throw could keep its text for both. + const rawName = safeRead(thrown, "name"); + const rawMessage = safeRead(thrown, "message"); + const rawStack = safeRead(thrown, "stack"); + const name = typeof rawName === "string" ? rawName : undefined; + const message = typeof rawMessage === "string" ? rawMessage : undefined; + const stack = typeof rawStack === "string" ? rawStack : undefined; + const error: Record = {}; + if (name) error.name = name; + if (message !== undefined) error.message = message; + if (stack) error.stack = stack; + // Custom properties go under `extra`, the same key and the same skip-list the + // bun/deno executors use for a failed child job, so an error carrying e.g. a + // `code` keeps it whether it failed as a task or as a step. + // + // Unlike the executors, which serialize once as the job dies, this record is + // stringified while the workflow is still running — once for the checkpoint + // POST and again for the wrapper's output. A property that cannot survive + // that would take the whole workflow down instead of reaching the `catch` the + // user wrote, so each one has to prove it round-trips before it is kept. + // `AxiosError.request` is the everyday example: it is circular via + // `socket._httpMessage`. Reading the property can throw too, since + // `getOwnPropertyNames` returns accessors and this invokes them. + // Objects only: on a primitive `Object.getOwnPropertyNames` returns its + // character indices, and `extra: {0: "b", 1: "o", …}` is noise the checkpoint + // would then have to carry. + if (thrown !== null && typeof thrown === "object") { + const extra: Record = {}; + // `ownKeys` is a proxy trap too, so even listing the properties can throw. + let keys: string[] = []; + try { + keys = Object.getOwnPropertyNames(thrown); + } catch { + keys = []; + } + for (const k of keys) { + if (SERIALIZED_ERROR_FIELDS.includes(k)) continue; + try { + // The snapshot, not the original: probing the original only proves it + // serialized *once*. A `toJSON` that succeeds and then throws — or one + // whose output depends on state that moves — would pass here and break + // the checkpoint encoding afterwards, when the failure has nowhere left + // to go. Keeping what the probe produced makes the check binding. + // `undefined` means the key would be omitted anyway (a function, a + // symbol, an explicit undefined), and `JSON.parse` rejects it for us. + extra[k] = JSON.parse(JSON.stringify(thrown[k])); + } catch { + // unreadable or unserializable: the failure itself still gets reported + } + } + if (Object.keys(extra).length > 0) error.extra = extra; + } + // `String(e)` throws in turn on a value with no `toString` to reach — + // `Object.create(null)`, a proxy that rejects the coercion — and this runs + // inside the catch that is reporting the user's failure, so an escape here + // replaces their error with an unrelated one and skips the checkpoint. + let fallback: string; + try { + fallback = String(e); + } catch { + fallback = `unrepresentable ${typeof e} thrown`; + } + return { + __wmill_error: true, + message: message ?? fallback, + step_key: key, + result: { error }, + }; +} + +/** @internal + * Rebuild the error a failed task or step throws. The run that produced the + * failure and every later replay go through here: a catch is control flow, + * `workflow()` re-runs its body from the top every round, so a handler that + * branches on the failure it caught must be handed the same thing in every + * round or it dispatches different tasks on the way back. */ +export function taskErrorFromMarker(marker: any, fallbackMessage: string): Error { + const err = new Error(marker?.message || fallbackMessage); + // Matches the python client, which raises TaskError here. Keeps a failed + // job's serialized error identical across the two languages. + err.name = "TaskError"; + (err as any).result = marker?.result; + (err as any).step_key = marker?.step_key; + (err as any).child_job_id = marker?.child_job_id; + return err; +}