fix(OMN-18978): name the stage cancelled by the delegation budget

contracts/OMN-18978.yaml

@@ -0,0 +1,33 @@
+---
+schema_version: "1.0.0"
+ticket_id: "OMN-18978"
+title: "A delegation budget cancellation names the stage it cancelled"
+summary: "Adds request-local typed dispatch progress to node_delegate_skill_orchestrator so a handler budget cancellation reports the innermost observed stage (bus subscribe, publish, terminal wait, local boot, inference, quality gate) in the timeout error, keeps it through cleanup, keeps concurrent requests separate and resets after dispatch. Tests: tests/unit/nodes/node_delegate_skill_orchestrator/test_timeout_stage_omn18978.py."
+is_seam_ticket: false
+interface_change: false
+interfaces_touched: []
+requirements:
+  - id: "req-timeout-names-stage"
+    statement: |-
+      A delegation cancelled by the handler budget must report the stage it was observed in when cancelled, keep that stage through cleanup, keep concurrent requests' stages separate, and leave an accepting dispatch's answer unchanged.
+    acceptance:
+      - id: "AC3"
+        statement: |-
+          AC3: a budget or port-owned timeout names its cancelled stage in the timeout error message on the bus and local paths, the cleanup stage never overwrites it, concurrent dispatches keep distinct stages, an accepting dispatch keeps its answer, and the progress context resets after dispatch -- falsifier: env -u PYTHONPATH uv run pytest tests/unit/nodes/node_delegate_skill_orchestrator/test_timeout_stage_omn18978.py -q
+evidence_requirements:
+  - kind: "tests"
+    description: "The timeout-stage tests pass at the head and fail at the merge base, where the dispatch progress module does not exist."
+    command: "env -u PYTHONPATH uv run pytest tests/unit/nodes/node_delegate_skill_orchestrator/test_timeout_stage_omn18978.py -q"
+emergency_bypass:
+  enabled: false
+  justification: ""
+  follow_up_ticket_id: ""
+dod_evidence:
+  - id: "dod-timeout-stage-ac3"
+    description: "The timeout-stage tests prove the cancelled stage is named, survives cleanup, stays per request and resets after dispatch."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/unit/nodes/node_delegate_skill_orchestrator/test_timeout_stage_omn18978.py -q"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC3"]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "omnimarket"
-version = "0.4.297"
+version = "0.4.298"
 description = "OmniMarket - Portable ONEX workflow package registry"
 authors = [
     {name = "OmniNode.ai", email = "contact@omninode.ai"}

src/omnimarket/nodes/node_delegate_skill_orchestrator/dispatch_progress.py

@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Keep cancellation evidence scoped to one delegation, including child tasks."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+from omnimarket.nodes.node_delegate_skill_orchestrator.models.model_delegation_dispatch_progress import (
+    DispatchStage,
+    ModelDelegationDispatchProgress,
+)
+
+current_dispatch_progress: ContextVar[ModelDelegationDispatchProgress | None] = (
+    ContextVar("delegation_dispatch_progress", default=None)
+)
+
+
+@contextmanager
+def dispatch_stage(stage: DispatchStage) -> Iterator[None]:
+    """Record the innermost cancelled stage before cleanup can change it."""
+    progress = current_dispatch_progress.get()
+    if progress is None:
+        yield
+        return
+    previous_stage = progress.stage
+    progress.stage = stage
+    try:
+        yield
+    except asyncio.CancelledError:
+        if progress.cancelled_stage is None:
+            progress.cancelled_stage = progress.stage
+        raise
+    finally:
+        progress.stage = previous_stage

src/omnimarket/nodes/node_delegate_skill_orchestrator/handlers/handler_delegate_skill.py

@@ -68,6 +68,9 @@
     ModelAttemptRubricVerdict,
 )
 from omnimarket.models.model_delegation_split_recombine import EnumDelegationSizeBand
+from omnimarket.nodes.node_delegate_skill_orchestrator.dispatch_progress import (
+    current_dispatch_progress,
+)
 from omnimarket.nodes.node_delegate_skill_orchestrator.models.model_delegate_skill_request import (
     ModelDelegateSkillRequest,
 )
@@ -80,6 +83,9 @@
     delegate_skill_terminal_from_response,
     resolve_terminal_failure_cause,
 )
+from omnimarket.nodes.node_delegate_skill_orchestrator.models.model_delegation_dispatch_progress import (
+    ModelDelegationDispatchProgress,
+)
 from omnimarket.nodes.node_delegate_skill_orchestrator.models.model_delegation_reap_context import (
     ModelDelegationReapContext,
 )
@@ -1173,6 +1179,8 @@ async def _dispatch_and_build_untagged_terminal(
                 execution_budget.terminal_delivery_margin_seconds
             ),
         )
+        progress = ModelDelegationDispatchProgress()
+        progress_token = current_dispatch_progress.set(progress)
         try:
             # OMN-15504: bound the AWAIT, not merely the code around it. The
             # dispatch is a single await, so there is no loop body in which a
@@ -1295,7 +1303,9 @@ async def _dispatch_and_build_untagged_terminal(
                 provenance=request.provenance,
                 error_message=(
                     f"delegation exceeded the handler execution budget of "
-                    f"{execution_timeout_seconds}s and was cancelled; the consumer commits "
+                    f"{execution_timeout_seconds}s and was cancelled at "
+                    f"stage={progress.cancelled_stage or progress.stage}; "
+                    "the consumer commits "
                     "this terminal instead of being evicted mid-handle "
                     f"(OMN-15504){queue_clause}"
                 ),
@@ -1326,6 +1336,9 @@ async def _dispatch_and_build_untagged_terminal(
                 budget_evidence=budget_evidence,
             )
 
+        finally:
+            current_dispatch_progress.reset(progress_token)
+
         return delegate_skill_terminal_from_response(
             _response_from_result(
                 request,

src/omnimarket/nodes/node_delegate_skill_orchestrator/models/model_delegation_dispatch_progress.py

@@ -0,0 +1,29 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Request-local evidence of the dispatch stage cancelled by the handler."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict
+
+DispatchStage = Literal[
+    "dispatch",
+    "subscribe",
+    "publish",
+    "terminal_wait",
+    "terminal_cleanup",
+    "effect_boot",
+    "inference",
+    "quality_gate",
+]
+
+
+class ModelDelegationDispatchProgress(BaseModel):
+    """Shared by one handler and its wait_for child, never by unrelated runs."""
+
+    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+
+    stage: DispatchStage = "dispatch"
+    cancelled_stage: DispatchStage | None = None

src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_local_delegation_dispatch.py

@@ -151,6 +151,10 @@
     ModelQualityGateResult,
     ModelQualityRuleEvaluation,
 )
+from omnimarket.nodes.node_delegate_skill_orchestrator.dispatch_progress import (
+    current_dispatch_progress,
+    dispatch_stage,
+)
 from omnimarket.nodes.node_delegate_skill_orchestrator.ports.evidence_db_resolution import (
     resolve_local_delegation_evidence_db,
 )
@@ -797,6 +801,9 @@ async def _run_effect_handler_with_killable_timeout(
         args=(effect_handler, request, result_queue),
         daemon=True,
     )
+    progress = current_dispatch_progress.get()
+    if progress is not None:
+        progress.stage = "effect_boot"
     boot_started = time.monotonic()
     process.start()
 
@@ -820,6 +827,8 @@ async def _run_effect_handler_with_killable_timeout(
                         timeout_seconds,
                     )
                     child_ready = True
+                    if progress is not None:
+                        progress.stage = "inference"
                     deadline = time.monotonic() + timeout_seconds
                     continue
 
@@ -2871,14 +2880,15 @@ async def _run_single_attempt(
         # worker can be terminated so ``asyncio.run`` has no orphaned thread to join.
         dispatch_deadline_seconds = timeout_seconds + _DISPATCH_TIMEOUT_BUFFER_SECONDS
         try:
-            if self._effect_process_boundary:
-                result = await _run_effect_handler_with_killable_timeout(
-                    self._effect_handler,
-                    call_request,
-                    timeout_seconds=dispatch_deadline_seconds,
-                )
-            else:
-                result = self._effect_handler(call_request)
+            with dispatch_stage("inference"):
+                if self._effect_process_boundary:
+                    result = await _run_effect_handler_with_killable_timeout(
+                        self._effect_handler,
+                        call_request,
+                        timeout_seconds=dispatch_deadline_seconds,
+                    )
+                else:
+                    result = self._effect_handler(call_request)
         except TimeoutError:
             failure_message = (
                 f"delegation call did not return within "
@@ -2960,46 +2970,47 @@ async def _run_single_attempt(
         #    refusal or empty answer returns success here but must NOT be recorded
         #    as a gate PASS. Resolve the task-class DoD checks from the routing
         #    authority and evaluate the deterministic verdict and graded score.
-        gate_result = await self._evaluate_quality_gate(
-            correlation_id=correlation_id,
-            task_type=task_type,
-            prompt=prompt,
-            grounding_source=stated_prompt,
-            content=gate_content
-            if gate_content is not None
-            else (result.content or ""),
-            quality_contract_mode=quality_contract_mode,
-            acceptance_criteria=acceptance_criteria,
-            # OMN-7942: the ALREADY-RESOLVED contract, not the caller's raw
-            # one. The gate resolves a task-class default of its own when
-            # handed None; passing the resolved value makes that a no-op and
-            # removes the second, independent resolution that could otherwise
-            # grade against a schema the model was not shown.
-            response_contract=effective_response_contract,
-            deliverable_evidence=ModelDelegationDeliverableEvidence(
-                output_shape=deliverable_contract.output_shape,
-                contract_sha256=canonical_deliverable_contract_sha256(
-                    deliverable_contract
+        with dispatch_stage("quality_gate"):
+            gate_result = await self._evaluate_quality_gate(
+                correlation_id=correlation_id,
+                task_type=task_type,
+                prompt=prompt,
+                grounding_source=stated_prompt,
+                content=gate_content
+                if gate_content is not None
+                else (result.content or ""),
+                quality_contract_mode=quality_contract_mode,
+                acceptance_criteria=acceptance_criteria,
+                # OMN-7942: the ALREADY-RESOLVED contract, not the caller's raw
+                # one. The gate resolves a task-class default of its own when
+                # handed None; passing the resolved value makes that a no-op and
+                # removes the second, independent resolution that could otherwise
+                # grade against a schema the model was not shown.
+                response_contract=effective_response_contract,
+                deliverable_evidence=ModelDelegationDeliverableEvidence(
+                    output_shape=deliverable_contract.output_shape,
+                    contract_sha256=canonical_deliverable_contract_sha256(
+                        deliverable_contract
+                    ),
+                    deliverable_sha256=hashlib.sha256(
+                        (result.content or "").encode()
+                    ).hexdigest(),
+                    deliverable_chars=len(result.content or ""),
+                    preamble_chars=extraction.preamble_chars,
+                    raw_chars=extraction.raw_chars,
+                    deliverable_start=extraction.deliverable_start,
+                    deliverable_end=extraction.deliverable_end,
                 ),
-                deliverable_sha256=hashlib.sha256(
-                    (result.content or "").encode()
-                ).hexdigest(),
-                deliverable_chars=len(result.content or ""),
-                preamble_chars=extraction.preamble_chars,
-                raw_chars=extraction.raw_chars,
-                deliverable_start=extraction.deliverable_start,
-                deliverable_end=extraction.deliverable_end,
-            ),
-            # OMN-18278: what the PROVIDER said about this response, not what
-            # the text says about itself. A response cut off by the output-token
-            # budget stopped mid-thought, so the model never emitted the
-            # terminator OMN-18379's segmenter cuts at and the scratchpad is the
-            # whole response — which every content heuristic reads as ordinary
-            # prose. The gate vetoes on this signal; without it the gate has no
-            # non-heuristic way to tell the two apart.
-            finish_reason=result.finish_reason,
-            reasoning_stripped_chars=result.reasoning_stripped_chars,
-        )
+                # OMN-18278: what the PROVIDER said about this response, not what
+                # the text says about itself. A response cut off by the output-token
+                # budget stopped mid-thought, so the model never emitted the
+                # terminator OMN-18379's segmenter cuts at and the scratchpad is the
+                # whole response — which every content heuristic reads as ordinary
+                # prose. The gate vetoes on this signal; without it the gate has no
+                # non-heuristic way to tell the two apart.
+                finish_reason=result.finish_reason,
+                reasoning_stripped_chars=result.reasoning_stripped_chars,
+            )
         from omnimarket.delegation.deliverable_extraction import (
             gate_result_with_output_refusal,
         )

src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_runtime_delegation_dispatch.py

@@ -20,6 +20,9 @@
     ModelDispatchBusTerminalResult,
 )
 from omnimarket.events.delegation import ModelDelegationRequest
+from omnimarket.nodes.node_delegate_skill_orchestrator.dispatch_progress import (
+    dispatch_stage,
+)
 from omnimarket.nodes.node_delegate_skill_orchestrator.models import (
     ModelRuntimeDelegationDispatchConfig,
 )
@@ -149,7 +152,8 @@ async def dispatch(
         )
 
         if not wait:
-            await self._publish_request(request)
+            with dispatch_stage("publish"):
+                await self._publish_request(request)
             return {
                 "status": "completed",
                 "content": "",
@@ -158,21 +162,25 @@ async def dispatch(
                 "quality_gate_passed": False,
             }
 
-        unsubscribe, queue = await self._subscribe_for_result(correlation_id)
+        with dispatch_stage("subscribe"):
+            unsubscribe, queue = await self._subscribe_for_result(correlation_id)
         try:
-            await self._publish_request(request)
+            with dispatch_stage("publish"):
+                await self._publish_request(request)
             timeout_seconds = float(self._config.wait_timeout_seconds)
-            terminal = await asyncio.wait_for(queue.get(), timeout=timeout_seconds)
+            with dispatch_stage("terminal_wait"):
+                terminal = await asyncio.wait_for(queue.get(), timeout=timeout_seconds)
         except TimeoutError:
             return {
                 "status": "timeout",
                 "error_message": (
                     f"timed out after {self._config.wait_timeout_seconds}s "
-                    "waiting for delegation result"
+                    "at stage=terminal_wait waiting for delegation result"
                 ),
             }
         finally:
-            await _unsubscribe(unsubscribe)
+            with dispatch_stage("terminal_cleanup"):
+                await _unsubscribe(unsubscribe)
 
         result: dict[str, object] = {
             "status": terminal.status,