fix: consolidated fix wave for resume, its events, TTL, and webhook pauses

Closes seven items from the final whole-branch review of resumable
executions:

1. Resume now claims the execution (status out of waiting, token and
   pausedNodeId cleared) via a version-checked compare-and-set write
   before running the resumed half, so a second concurrent resume with
   the same token fails instead of replaying the run.
2. ResumeExecution now builds and passes the same event callback
   ExecuteWorkflow uses, so a resumed run emits node.* and execution.*
   events again.
3. A second pause on an already-stored execution now goes through a
   remove-then-insert so its longer TTL actually lands, instead of an
   update() that cannot carry one.
4. The resume endpoint now requires "approved" explicitly and type-checks
   approved/data/token before reading them, rejecting bad input with 400
   instead of a default approval or a generic 500.
5. A Waiting execution now releases its scheduler slot, and the duplicate
   execution.waiting emission is collapsed into one with the fuller shape.
6. A webhook-triggered workflow that pauses now returns 202 with a
   waiting status and broadcasts .waiting instead of .completed; docs
   updated to match.
7. Added a fixture covering _webhookResponse set from inside a loop body.

Full fixture suite: 37/37 (36 existing + the new one).

docs/nodes.md

@@ -236,8 +236,10 @@ Two limits are deliberate. A pause inside a Loop body cannot be resumed, because
 loop iteration state is not part of the stored execution, so a node must refuse
 to pause there rather than record something unanswerable. And a webhook cannot
 wait for an approval - the HTTP request is still open and its deadline is
-35 seconds - so a webhook-triggered workflow that pauses returns immediately
-with the execution id.
+35 seconds - so a webhook-triggered workflow that pauses returns a 202 with
+`{ "executionId": "...", "status": "waiting" }` rather than blocking for an
+answer that has nowhere to arrive on this connection. Answer it the same way
+as any other paused execution, with `POST /api/v1/executions/{id}/resume`.
 
 ## Answering a Webhook
 

src/runner/runner_service.cpp

@@ -172,7 +172,30 @@ grpc::Status RunnerServiceImpl::ResumeExecution(grpc::ServerContext* context,
         }
     }
 
-    auto result = engine_.resume(request->execution_id(), request->token(), payload);
+    // The resumed half of the run needs the same event plumbing the initial
+    // half gets in ExecuteWorkflow above, or the UI shows nothing for it and
+    // execution.failed never reaches the handler that runs error workflows.
+    // ExecuteWorkflow's callback stamps workflowId onto every event because
+    // most of the engine's per-node events don't carry it themselves; this
+    // does the same, reading workflowId from the execution record up front
+    // since resume() (unlike execute()) is not handed a parsed Workflow by
+    // its caller.
+    std::string workflow_id;
+    auto stored = storage_.get("executions", request->execution_id());
+    if (stored.ok()) {
+        workflow_id = stored.value().value("workflowId", "");
+    }
+
+    ExecutionCallback callback;
+    if (event_callback_) {
+        callback = [this, workflow_id](const std::string& event_type, const nlohmann::json& data) {
+            nlohmann::json event_data = data;
+            event_data["workflowId"] = workflow_id;
+            event_callback_(event_type, event_data);
+        };
+    }
+
+    auto result = engine_.resume(request->execution_id(), request->token(), payload, callback);
     if (result.failed()) {
         return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, result.error().message());
     }

src/runner/workflow_engine.cpp

@@ -706,13 +706,11 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
                 auto& stored_node = result.node_results[node_id];
                 stored_node.output.erase("_pause");
 
-                if (callback) {
-                    callback("execution.waiting", {
-                        {"executionId", result.execution_id},
-                        {"nodeId", node_id},
-                        {"expiresAt", result.pause_expires_at}
-                    });
-                }
+                // No callback here - the generic terminal callback below (after
+                // storeExecution) already emits "execution.waiting" once this
+                // pass ends, and it carries nodeId/expiresAt too. Emitting here
+                // as well produced the same event twice with two different
+                // payload shapes.
 
                 LOG_INFO("Execution {} paused at node {}", result.execution_id, node_id);
                 break;
@@ -816,13 +814,23 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
         // error workflow, a notification - needs to know what went wrong, and
         // this event is the first one out, so omitting it left listeners with a
         // failure and no reason for it.
-        callback("execution." + executionStatusToString(result.status), {
+        nlohmann::json event_data = {
             {"executionId", result.execution_id},
             {"workflowId", result.workflow_id},
             {"status", executionStatusToString(result.status)},
             {"error", result.error},
             {"output", truncateLargeValues(result.final_output)}
-        });
+        };
+        // A Waiting result is the one terminal status that carries reader-
+        // relevant fields the generic shape above doesn't have - which node
+        // is asking, and when the wait itself expires. This is the only
+        // "execution.waiting" emission (the pause block above deliberately
+        // does not emit its own), so those fields land here.
+        if (result.status == ExecutionStatus::Waiting) {
+            event_data["nodeId"] = result.paused_node_id;
+            event_data["expiresAt"] = result.pause_expires_at;
+        }
+        callback("execution." + executionStatusToString(result.status), event_data);
     }
 
     LOG_INFO("Workflow execution {} completed with status: {}",
@@ -871,6 +879,42 @@ common::Result<ExecutionResult> WorkflowEngine::resume(const std::string& execut
             "Execution " + execution_id + " has no usable workflow snapshot to resume against");
     }
 
+    // Claim the execution before running it. Everything above this point only
+    // reads the record - two concurrent resumes with the same token both pass
+    // every check above and would otherwise both call execute() on the second
+    // half of the run, producing duplicate side effects (a resumed run is not
+    // idempotent - it sends emails, charges cards, posts messages). The write
+    // below moves status out of "waiting" and clears the token/pausedNodeId so
+    // a second resume cannot match either the status check or the token check.
+    // storage_.update() with a non-zero expected_version goes through upstream
+    // updateIfVersion(), a genuine server-side compare-and-set (verified against
+    // smartbotic-database's client.cpp: it sends expected_version on the wire
+    // and the server rejects a stale write), so a racing second resume loses
+    // this write and its execute() call never happens.
+    const int64_t expected_version = record.value("_version", static_cast<int64_t>(0));
+    if (expected_version <= 0) {
+        // Every document returned by get() carries a _version metadata field
+        // stamped by the database on every write. Its absence means either a
+        // very old record predating that guarantee or something reading the
+        // record incorrectly - either way, resuming without a real
+        // compare-and-set would silently reopen the replay window this claim
+        // exists to close, so refuse rather than proceed unprotected.
+        return common::Error(common::ErrorCode::Internal,
+            "Execution " + execution_id + " has no version metadata to claim it safely with");
+    }
+
+    nlohmann::json claim = record;
+    claim.erase("_version");
+    claim["status"] = "resuming";
+    claim["pauseToken"] = "";
+    claim["pausedNodeId"] = "";
+
+    auto claim_result = storage_.update("executions", execution_id, claim, expected_version);
+    if (claim_result.failed()) {
+        return common::Error(common::ErrorCode::FailedPrecondition,
+            "Execution " + execution_id + " is already being resumed");
+    }
+
     Workflow workflow = Workflow::fromJson(record["workflowSnapshot"]);
 
     // Defend against snapshots stored before id/name/settings were added to
@@ -1535,6 +1579,36 @@ void WorkflowEngine::storeExecution(const ExecutionResult& result) {
     // update when it is already there.
     auto existing = storage_.get("executions", result.execution_id);
     if (existing.ok()) {
+        if (result.status == ExecutionStatus::Waiting) {
+            // storage_.update() has no ttl_ms parameter - only insert() can set
+            // one. A second pause on an already-existing record recomputes
+            // ttl_ms above from its own (possibly much later) pause_expires_at,
+            // but a plain update() would leave the record on whatever TTL its
+            // very first insert got, undoing the floor above for every pause
+            // after the first - exactly the silent-data-loss shape called out
+            // elsewhere on this branch. Remove and reinsert so the fresh
+            // deadline actually lands. This opens a narrow window between the
+            // two calls where a process crash mid-write would lose the record
+            // outright (the remove lands, the insert never runs); accepted
+            // here because it only applies to a Waiting execution, the window
+            // is a single pair of adjacent calls rather than a standing gap,
+            // and the alternative is a certainty rather than a small chance:
+            // every long-lived second approval would otherwise expire early.
+            auto removed = storage_.remove("executions", result.execution_id);
+            if (removed.failed()) {
+                LOG_ERROR("Failed to remove execution {} before re-inserting with a fresh TTL: {}",
+                          result.execution_id, removed.error().message());
+            }
+            auto reinserted = storage_.insert("executions", result.toJson(), result.execution_id, ttl_ms);
+            if (reinserted.failed()) {
+                LOG_ERROR("Failed to re-store waiting execution {} with a fresh TTL: {}",
+                          result.execution_id, reinserted.error().message());
+            } else {
+                LOG_DEBUG("Re-stored waiting execution {} with a fresh TTL", result.execution_id);
+            }
+            return;
+        }
+
         auto updated = storage_.update("executions", result.execution_id, result.toJson());
         if (updated.failed()) {
             LOG_ERROR("Failed to update execution {}: {}", result.execution_id, updated.error().message());

src/webserver/api/execution_controller.cpp

@@ -161,14 +161,45 @@ void ExecutionController::resumeExecution(const httplib::Request& req, httplib::
         return;
     }
 
+    if (!body.is_object()) {
+        sendError(res, "Body must be a JSON object", 400);
+        return;
+    }
+
+    // Everything below reads body.value(...), which throws json::type_error
+    // when the key exists but holds the wrong type - that escaped the parse
+    // try/catch above and became a generic 500, the same defect fixed earlier
+    // on this branch at spec.value("status", 200). Each field is type-checked
+    // before it is read, and rejected with 400 naming which one is wrong.
+
+    if (body.contains("token") && !body["token"].is_string()) {
+        sendError(res, "token must be a string", 400);
+        return;
+    }
     const std::string token = body.value("token", "");
     if (token.empty()) {
         sendError(res, "A token is required to answer a paused execution", 400);
         return;
     }
 
+    if (body.contains("data") && !body["data"].is_object()) {
+        sendError(res, "data must be an object", 400);
+        return;
+    }
     nlohmann::json payload = body.value("data", nlohmann::json::object());
-    payload["approved"] = body.value("approved", true);
+
+    // This is an approval endpoint. Defaulting a missing or misspelled
+    // "approved" to true would silently approve whatever it is guarding -
+    // require it explicitly rather than assume yes.
+    if (!body.contains("approved")) {
+        sendError(res, "approved is required (true or false)", 400);
+        return;
+    }
+    if (!body["approved"].is_boolean()) {
+        sendError(res, "approved must be true or false", 400);
+        return;
+    }
+    payload["approved"] = body["approved"].get<bool>();
     payload["answeredBy"] = ctx.user_id;
 
     auto runner = load_balancer_.selectRunner();
@@ -282,9 +313,14 @@ void ExecutionController::receiveExecutionEvent(const httplib::Request& req, htt
 
         // Release the scheduler slot held by this run. Scheduled dispatch is
         // fire-and-forget, so these events are the only signal that a run ended.
+        // A Waiting execution is no longer occupying the runner either - it is
+        // parked on a person, not running - so it releases the slot too.
+        // Without this, "Schedule -> ... -> Wait for Approval" holds its slot
+        // until the watchdog deadline expires and logs a misleading "exceeded
+        // its deadline" for a run that was correctly waiting for an answer.
         if (!execution_id.empty() &&
             (event_type == "execution.completed" || event_type == "execution.failed" ||
-             event_type == "execution.cancelled")) {
+             event_type == "execution.cancelled" || event_type == "execution.waiting")) {
             scheduler_.notifyExecutionFinished(workflow_id, execution_id);
         }
 

src/webserver/api/webhook_controller.cpp

@@ -163,6 +163,24 @@ void WebhookController::handleWebhook(const httplib::Request& req, httplib::Resp
         return;
     }
 
+    // A workflow that paused mid-run is not done: final_output is still null
+    // (the body would just be the literal string "null"), and labelling this
+    // ".completed" is false. Report it honestly - 202 Accepted, a status of
+    // "waiting", and a ".waiting" broadcast - and return before the completed
+    // path below, which assumes the run actually finished.
+    if (grpc_res.status() == proto::EXECUTION_STATUS_WAITING) {
+        ws_server_.broadcast("executions." + grpc_res.execution_id() + ".waiting", {
+            {"executionId", grpc_res.execution_id()},
+            {"workflowId", workflow_id},
+            {"status", "waiting"}
+        });
+        sendJson(res, {
+            {"executionId", grpc_res.execution_id()},
+            {"status", "waiting"}
+        }, 202);
+        return;
+    }
+
     // Broadcast execution event
     ws_server_.broadcast("executions." + grpc_res.execution_id() + ".completed", {
         {"executionId", grpc_res.execution_id()},

tests/nodes/respond-to-webhook-loop.json

@@ -0,0 +1,24 @@
+{
+  "name": "verify-respond-to-webhook-loop",
+  "nodes": [
+    {"id": "n1", "name": "Webhook", "type": "post-trigger", "position": {"x": 0, "y": 0}, "config": {}},
+    {"id": "items", "name": "Items", "type": "code", "position": {"x": 0, "y": 100},
+     "config": {"code": "return { items: ['alpha', 'beta', 'gamma'] };"}},
+    {"id": "loop", "name": "Loop", "type": "loop", "position": {"x": 0, "y": 200},
+     "config": {"inputField": "data.result.items"}},
+    {"id": "respond", "name": "Respond", "type": "respond-to-webhook", "position": {"x": 0, "y": 300},
+     "config": {"status": 200, "bodySource": "input", "bodyField": "item", "contentType": "text/plain"}}
+  ],
+  "connections": [
+    {"sourceNodeId": "n1", "sourceOutput": "main", "targetNodeId": "items", "targetInput": "data"},
+    {"sourceNodeId": "items", "sourceOutput": "main", "targetNodeId": "loop", "targetInput": "data"},
+    {"sourceNodeId": "loop", "sourceOutput": "loop", "targetNodeId": "respond", "targetInput": "data"}
+  ],
+  "http": {
+    "method": "POST",
+    "path": "/webhook/{workflowId}",
+    "body": {"hello": "world"},
+    "expectStatus": 200,
+    "expectBodyContains": ["gamma"]
+  }
+}