Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
475 changes: 462 additions & 13 deletions src/omnibase_infra/cli/cli_delegate.py

Large diffs are not rendered by default.

43 changes: 43 additions & 0 deletions src/omnibase_infra/cli/delegate_terminal_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,10 +59,17 @@
)

__all__ = [
"REPLY_TIMEOUT_WORKFLOW_RESULT",
"DelegateReplyTimeoutError",
"DelegateTerminalUnresolvedError",
"is_reply_timeout",
"reply_timeout_error",
"resolve_delegate_terminal",
]

#: The runtime's own workflow result for a wait that ended with no terminal.
REPLY_TIMEOUT_WORKFLOW_RESULT = "timeout"


class DelegateTerminalUnresolvedError(ValueError):
"""A delegate receipt carried no resolvable delegation terminal.
Expand All @@ -73,6 +80,42 @@ class DelegateTerminalUnresolvedError(ValueError):
"""


class DelegateReplyTimeoutError(DelegateTerminalUnresolvedError):
"""A published delegation command got no terminal before the wait ended (OMN-20386).

A run the runtime ended as ``timeout`` that recorded a wire correlation id
published its command and heard nothing back; one without a wire
correlation id never published and is a ``DelegatePrePublishFailureError``.
Subclasses :class:`DelegateTerminalUnresolvedError` so every existing catch
site keeps catching it, while its message names the cause, the run id and
the wire correlation id instead of describing a receipt that "carries no
resolvable terminal". That error is a ``ValueError`` and ``delegate_command``
turns every ``ValueError`` into ``click.UsageError``: exit 2 and a
``Usage:`` banner, which is what a lane reply timeout used to produce.
"""


def is_reply_timeout(summary: dict[str, object]) -> bool:
"""Whether a runtime summary records a published command that got no reply."""
return bool(
str(summary.get("workflow_result") or "") == REPLY_TIMEOUT_WORKFLOW_RESULT
and summary.get("wire_correlation_id")
)


def reply_timeout_error(envelope: dict[str, object]) -> DelegateReplyTimeoutError:
"""The error for a receipt that :func:`is_reply_timeout` accepted."""
result = envelope.get("result")
summary = result if isinstance(result, dict) else {}
return DelegateReplyTimeoutError(
f"delegate run {envelope.get('run_id')} ended with cause "
f"{REPLY_TIMEOUT_WORKFLOW_RESULT}: the command was published (wire "
f"correlation id {summary.get('wire_correlation_id')}) and the receipt "
"carries no resolvable delegation terminal, because none arrived before "
"the wait ended"
)


def _absent_fields(error: ValidationError) -> tuple[str, ...]:
"""Name the required fields a carrier did not supply, in report order.

Expand Down
22 changes: 20 additions & 2 deletions src/omnibase_infra/cli/receipt_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -988,6 +988,7 @@ def run_receipt_mode(
locus_decision: ModelDelegateLocusDecision | None = None,
receipt_renderer: Callable[[object], bool] | None = None,
phase_stopwatch: DelegatePhaseStopwatch | None = None,
run_id: uuid.UUID | None = None,
) -> int:
"""Serialize receipt mode and restore its process-global state on exit."""
with _RECEIPT_MODE_LOCK_STATE.lock:
Expand Down Expand Up @@ -1015,6 +1016,7 @@ def run_receipt_mode(
locus_decision=locus_decision,
receipt_renderer=receipt_renderer,
phase_stopwatch=phase_stopwatch,
run_id=run_id,
)
except BaseException as exc:
operation_error = exc
Expand Down Expand Up @@ -1073,6 +1075,7 @@ def _run_receipt_mode(
locus_decision: ModelDelegateLocusDecision | None = None,
receipt_renderer: Callable[[object], bool] | None = None,
phase_stopwatch: DelegatePhaseStopwatch | None = None,
run_id: uuid.UUID | None = None,
) -> int:
"""Execute the node and print exactly one ``ModelSkillResult`` JSON.

Expand Down Expand Up @@ -1108,11 +1111,18 @@ def _run_receipt_mode(
bus reports connect, subscribe, publish and the wait for the terminal to
it; ``None`` (every other caller) builds the plain runtime, unchanged.

``run_id`` is the identity of this run when the caller minted one; ``None``
mints a fresh one.

Returns the process exit code (the runtime's exit code; 1 when the
runtime raised before producing a workflow result).
"""
state_root = state_root.resolve()
run_id = uuid.uuid4()
# OMN-17427: a caller that already told someone this run's id (``onex
# delegate`` prints it and files a provisional receipt under it before the
# wait) hands it in, so the run directory the caller named is the one the
# final receipt lands in. ``None`` keeps the fresh id every other caller gets.
run_id = run_id if run_id is not None else uuid.uuid4()
# One correlation id shared by the skill-started / skill-completed pair so
# the skill_executions projection can join the two rows (OMN-13830). This is
# decided BEFORE the run because the started event fires before the body.
Expand Down Expand Up @@ -1470,7 +1480,15 @@ def _run_receipt_mode(
validation_error = receipt_validator(receipt)
if validation_error is not None:
summary = ModelReceiptRuntimeSummary(
workflow_result="error",
# OMN-20386: a published command the lane never answered keeps
# the runtime's own word for it. Rewriting it as ``error`` would
# take the cause off the receipt the caller reads.
workflow_result=(
workflow_result.value
if workflow_result is EnumWorkflowResult.TIMEOUT
and workflow_data.get("wire_correlation_id")
else "error"
),
exit_code=1,
workflow=str(contract_path),
handler_locus=_json_str(workflow_data.get("handler_locus")),
Expand Down
38 changes: 38 additions & 0 deletions src/omnibase_infra/enums/enum_delegate_incomplete_class.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
# SPDX-License-Identifier: MIT
"""How a delegation that left no terminal ended, as its receipt reports it (OMN-17427).

``onex delegate`` writes ``receipt.json`` for every way a run can end. Most
write a terminal; these four are the ones that end without one, and the
receipt's ``terminal_class`` says which.

.. versionadded:: OMN-17427
"""

from __future__ import annotations

from enum import Enum

__all__ = ["EnumDelegateIncompleteClass"]


class EnumDelegateIncompleteClass(str, Enum):
"""The class of a receipt written for a run with no delegation terminal."""

IN_FLIGHT = "in_flight"
"""Filed before the terminal wait; what a caller's SIGKILL leaves behind.

Not an outcome: a finished process replaces it, and one found after its
process has gone means the run ended without a terminal.
"""

TIMEOUT = "timeout"
"""The wait for the terminal ended at this CLI's own bound, or the runtime
recorded ``timeout`` and no terminal payload."""

INTERRUPTED = "interrupted"
"""The caller ended the process with a signal it can be answered on
(``SIGTERM``, ``SIGHUP``, ``SIGINT``): its own timeout, or Ctrl-C."""

FAILED = "failed"
"""The run ended and its receipt holds no terminal payload to write from."""
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
pre_publish_failure_from_receipt,
)
from omnibase_infra.cli.delegate_terminal_resolver import (
DelegateReplyTimeoutError,
DelegateTerminalUnresolvedError,
)
from omnibase_infra.cli.model_receipt_runtime_summary import (
Expand Down Expand Up @@ -300,6 +301,9 @@ def test_published_run_without_a_terminal_keeps_the_terminal_sentence(
_delegation_result(envelope)
assert not isinstance(raised.value, DelegatePrePublishFailureError)
assert _TERMINAL_SENTENCE in str(raised.value)
# OMN-20386: and it names the cause, so the caller is not left to infer it.
assert isinstance(raised.value, DelegateReplyTimeoutError)
assert "cause timeout" in str(raised.value)

def test_completed_run_without_a_terminal_is_not_pre_publish(self) -> None:
"""OMN-18569's completed-but-empty shape keeps failing closed as before."""
Expand Down
10 changes: 10 additions & 0 deletions tests/integration/cli/test_delegate_refusal_receipt_omn19006.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@

import ast
import json
import signal
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
Expand All @@ -45,6 +46,7 @@
from omnibase_infra.backends.auto_configure import EventBusResolutionAmbiguousError
from omnibase_infra.cli import cli_delegate
from omnibase_infra.cli.cli_delegate import (
DelegateCallerInterruptedError,
DelegateCommand,
DelegateTimeoutExceededError,
delegate_command,
Expand Down Expand Up @@ -368,6 +370,14 @@ class BranchCase:
),
},
),
"caller-interrupted": BranchCase(
covers="except DelegateCallerInterruptedError",
args=lambda tmp: _common(tmp),
reason="ended by SIGTERM",
patches={
"run_receipt_mode": _raises(DelegateCallerInterruptedError(signal.SIGTERM)),
},
),
"exit-without-a-receipt": BranchCase(
covers=_KEY_SYS_EXIT,
args=lambda tmp: _common(tmp),
Expand Down
25 changes: 18 additions & 7 deletions tests/integration/cli/test_dispatched_run_files_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,22 +182,33 @@ def _receipt_without_a_terminal(
def test_an_unresolvable_terminal_refuses_and_names_the_field(
self, tmp_path: Path
) -> None:
receipt = self._receipt_without_a_terminal(
workflow=(
"/site-packages/omnimarket/nodes/"
"node_delegate_skill_orchestrator/contract.yaml"
)
)
with pytest.raises(DelegateTerminalUnresolvedError) as raised:
_write_local_run_files(
receipt=self._receipt_without_a_terminal(
workflow=(
"/site-packages/omnimarket/nodes/"
"node_delegate_skill_orchestrator/contract.yaml"
)
),
receipt=receipt,
state_root=tmp_path,
addressing=_ADDRESSING,
prompt=_PROMPT,
task_type="summarization",
task_type_resolution="explicit",
)
assert "attempts" in str(raised.value)
assert not (tmp_path / "runs").exists()
# OMN-17427: the refusal still raises and still names the field. What
# changed is that the run is no longer left with nothing on disk: a
# ``failed`` receipt carries the same cause.
written = json.loads(
(tmp_path / "runs" / str(receipt.run_id) / "receipt.json").read_text(
encoding="utf-8"
)
)
assert written["terminal_class"] == "failed"
assert written["terminal_recorded"] is False
assert "attempts" in written["failure_reason"]

def test_another_nodes_run_is_still_silent(self, tmp_path: Path) -> None:
"""The refusal did not widen. Negative control on the class above."""
Expand Down
9 changes: 8 additions & 1 deletion tests/unit/cli/test_cli_delegate.py
Original file line number Diff line number Diff line change
Expand Up @@ -2436,7 +2436,14 @@ def test_absent_terminal_raises_naming_both_carrier_fields(
message = str(raised.value)
assert "terminal_payload: absent" in message
assert "handler_result: absent" in message
assert not (tmp_path / "runs").exists()
# OMN-17427: the refusal still raises, and the run is no longer left
# without a receipt -- the receipt carries the same cause.
written = json.loads(
(tmp_path / "runs" / str(receipt.run_id) / "receipt.json").read_text()
)
assert written["terminal_class"] == "failed"
assert written["terminal_recorded"] is False
assert "terminal_payload: absent" in written["failure_reason"]

def test_terminal_of_an_unrecognised_shape_names_the_absent_field(
self, tmp_path: Path
Expand Down
Loading
Loading