omnigent-ai--omnigent/tests/tools/builtins/test_timer.py

"""
Unit tests for ``sys_timer_set`` and ``sys_timer_cancel`` (step 10
of the harness contract migration).

Covers the schema shape, argument validation, and ToolManager
registration gating — paths that don't require a running DBOS
workflow. The actual firing behavior (sleep + send +
auto-rendering) requires DBOS and is exercised by the server
integration suite.

Mirrors the structure of :mod:`tests.tools.builtins.test_async_inbox`,
which gates ``sys_call_async`` / ``sys_read_inbox`` /
``sys_cancel_async`` on ``async_enabled``. The timer family is
gated on ``timers``.
"""

from __future__ import annotations

import json

import pytest

from omnigent.spec import AgentSpec
from omnigent.tools.base import ToolContext
from omnigent.tools.builtins.timer import (
    SysTimerCancelTool,
    SysTimerSetTool,
    validate_timer_set_args,
)
from omnigent.tools.manager import ToolManager

# Stub :class:`ToolContext` for the validation paths below. All the
# parametrized invalid-argument tests hit early-exit error branches
# in ``SysTimerSetTool.invoke`` BEFORE the conversation_id check, so
# the conversation_id value here doesn't affect them. The dedicated
# ``test_set_missing_conversation_id_returns_error`` test exercises
# the conversation_id branch with a different stub.
_STUB_CTX = ToolContext(task_id="task_parent", agent_id="agent_x", conversation_id="conv_x")
_STUB_CTX_NO_CONV = ToolContext(task_id="task_parent", agent_id="agent_x", conversation_id=None)


# ─── Schema shape ────────────────────────────────────────────


def test_set_schema_required_fields_and_no_extras() -> None:
    """
    ``sys_timer_set`` requires ``seconds`` and rejects unknown
    properties. ``repeat`` and ``note`` are optional.

    A regression that loosened the required list would let the LLM
    schedule a timer with no delay (the workflow would then fire
    immediately in a tight loop for repeat=true). Allowing extras
    would silently drop unknown keys instead of failing fast.
    """
    schema = SysTimerSetTool().get_schema()["function"]["parameters"]
    assert schema["required"] == ["seconds"]
    assert schema["additionalProperties"] is False
    assert set(schema["properties"].keys()) == {"seconds", "repeat", "note"}
    assert schema["properties"]["repeat"]["default"] is False


def test_cancel_schema_required_fields_and_no_extras() -> None:
    """
    ``sys_timer_cancel`` requires ``timer_id`` and rejects unknown
    properties.

    Without the required field, the LLM could call cancel with no
    args; the validation branch returns an error string but a
    schema-enforced reject is the cleaner front line.
    """
    schema = SysTimerCancelTool().get_schema()["function"]["parameters"]
    assert schema["required"] == ["timer_id"]
    assert schema["additionalProperties"] is False
    assert set(schema["properties"].keys()) == {"timer_id"}


def test_tools_are_synchronous() -> None:
    """
    Both timer tools return ``is_async() == False``.

    The CALL is synchronous (the LLM gets the timer_id back
    immediately as the tool result), even though the FIRING is
    asynchronous and arrives later as a persisted conversation
    item. A regression where ``is_async`` flipped to ``True``
    would route ``invoke`` to ``dispatch_async`` and the LLM
    would see a "task in progress" handle JSON instead of the
    timer_id it needs to later cancel. The cross-harness PATCH
    flow also doesn't propagate string returns from
    ``dispatch_async`` back to all harness types, so the sync
    ``invoke`` path is what produces a clean function_call_output
    in the conversation.
    """
    assert SysTimerSetTool().is_async() is False
    assert SysTimerCancelTool().is_async() is False


# ─── SysTimerSetTool argument validation ─────────────────────


@pytest.mark.parametrize(
    "args_json,expected_error_substring",
    [
        # Missing required field — passes empty object so the JSON
        # parser succeeds and the missing-seconds branch fires.
        ("{}", "seconds must be a number"),
        # Negative seconds — the underlying DBOS sleep would raise
        # at workflow time, but rejecting at the tool boundary
        # surfaces the error to the LLM with a clearer message.
        ('{"seconds": -1}', "seconds must be non-negative"),
        # Above the cap — guards against an LLM hallucination
        # parking a workflow indefinitely.
        ('{"seconds": 10000000}', "seconds must be <="),
        # Bool gets rejected explicitly because Python treats
        # ``isinstance(True, int)`` as True; without the bool
        # check, ``True`` would silently coerce to ``1.0``.
        ('{"seconds": true}', "seconds must be a number"),
        # ``repeat`` MUST be a real bool, not a truthy string —
        # YAML / LLM JSON often confuses these.
        ('{"seconds": 1, "repeat": "yes"}', "repeat must be a boolean"),
        # ``note`` MUST be a string when present.
        ('{"seconds": 1, "note": 5}', "note must be a string"),
    ],
)
def test_set_invalid_args_return_error(args_json: str, expected_error_substring: str) -> None:
    """
    Each malformed argument shape produces a structured ``{"error":
    ...}`` response without starting a workflow.

    Without these checks, a bad call would either crash inside the
    workflow (LLM sees an opaque "internal error") or silently
    coerce wrong types. The error path is the public surface for
    invalid input — must remain stable.
    """
    result_json = SysTimerSetTool().invoke(args_json, _STUB_CTX)
    result = json.loads(result_json)
    assert "error" in result, f"expected error key in {result!r}"
    assert expected_error_substring in result["error"]


def test_set_malformed_json_returns_parse_error() -> None:
    """
    Malformed JSON produces a structured ``{"error": "invalid
    arguments: ..."}`` response, not a 500-style crash.

    Mirrors :class:`SysCallAsyncTool`'s own JSON-decode error
    handling — the tool is on the LLM-facing boundary and any
    parse failure must round-trip as an error string.
    """
    result_json = SysTimerSetTool().invoke("{not json", _STUB_CTX)
    result = json.loads(result_json)
    assert "error" in result
    assert "invalid arguments" in result["error"]


def test_set_missing_conversation_id_returns_error() -> None:
    """
    Valid args + ``ctx.conversation_id is None`` returns a structured
    error (no workflow is started).

    The timer workflow appends firings to the conversation store,
    so it MUST have a destination conversation. Without that, the
    firings would have nowhere to land — the tool fails loud here
    rather than silently dropping them. A regression that omitted
    this guard would surface much later as a workflow-time
    exception or, worse, silently lost firings.
    """
    result_json = SysTimerSetTool().invoke('{"seconds": 1, "note": "x"}', _STUB_CTX_NO_CONV)
    result = json.loads(result_json)
    assert "error" in result
    assert "conversation" in result["error"].lower()


def test_set_valid_args_in_process_reports_no_schedule() -> None:
    """
    Fully valid args off the runner dispatch path return a structured
    error, never a raise or a false success.

    The runner intercepts ``sys_timer_set`` before this builtin is
    reached (see ``tool_dispatch._execute_timer_set``). This guards the
    in-process fallback: a future non-runner caller must fail cleanly
    with no ``timer_id`` (there is no scheduled timer to cancel), rather
    than hit the old ``NotImplementedError`` trap.
    """
    result_json = SysTimerSetTool().invoke('{"seconds": 5, "note": "x"}', _STUB_CTX)
    result = json.loads(result_json)
    assert "error" in result
    assert "timer_id" not in result
    assert result.get("status") != "scheduled"


# ─── Shared validation helper ────────────────────────────────


def test_validate_timer_set_args_accepts_valid_shapes() -> None:
    """
    ``validate_timer_set_args`` returns ``(seconds, repeat, note)`` for
    valid input — the tuple the runner firing loop and the builtin both
    consume.

    This is the single contract both surfaces share; a drift here would
    desync the runner path from the LLM-facing schema.
    """
    assert validate_timer_set_args({"seconds": 5}) == (5.0, False, None)
    assert validate_timer_set_args({"seconds": 5, "repeat": True, "note": "x"}) == (
        5.0,
        True,
        "x",
    )


@pytest.mark.parametrize(
    "args,expected_error_substring",
    [
        ({}, "seconds must be a number"),
        ({"seconds": -1}, "seconds must be non-negative"),
        ({"seconds": 10_000_000}, "seconds must be <="),
        ({"seconds": True}, "seconds must be a number"),
        ({"seconds": 1, "repeat": "yes"}, "repeat must be a boolean"),
        ({"seconds": 1, "note": 5}, "note must be a string"),
    ],
)
def test_validate_timer_set_args_rejects_bad_shapes(
    args: dict[str, object], expected_error_substring: str
) -> None:
    """
    Each malformed shape yields an error message (a ``str``), not a
    tuple — the same messages both timer surfaces return to the LLM.
    """
    result = validate_timer_set_args(args)
    assert isinstance(result, str)
    assert expected_error_substring in result


# ─── SysTimerCancelTool argument validation ──────────────────


def test_cancel_missing_timer_id_returns_error() -> None:
    """
    Missing ``timer_id`` returns ``{"error": "timer_id is required"}``.

    The tool can't address a workflow without an id — the cancel
    has nothing to act on. Returning an error is preferred over
    a silent ``not_found`` because the latter would mask LLM bugs
    where it forgot to thread the timer_id through.
    """
    result_json = SysTimerCancelTool().invoke("{}", _STUB_CTX)
    result = json.loads(result_json)
    assert result == {"error": "timer_id is required"}


def test_cancel_empty_string_timer_id_returns_error() -> None:
    """
    Empty-string ``timer_id`` is rejected with the same error as a
    missing key.

    Distinct empty-string handling matters because the LLM
    sometimes passes ``""`` for omitted fields rather than
    omitting them — without this branch, an empty string would
    flow into ``get_workflow_status`` and produce a confusing
    DBOS-internal error.
    """
    result_json = SysTimerCancelTool().invoke('{"timer_id": ""}', _STUB_CTX)
    result = json.loads(result_json)
    assert result == {"error": "timer_id is required"}


def test_cancel_malformed_json_returns_parse_error() -> None:
    """Malformed JSON produces a parse-error response on the cancel tool too."""
    result_json = SysTimerCancelTool().invoke("not json", _STUB_CTX)
    result = json.loads(result_json)
    assert "error" in result
    assert "invalid arguments" in result["error"]


# ─── ToolManager registration gating ─────────────────────────


def test_timers_false_does_not_register() -> None:
    """
    With ``timers=False`` (the default) the manager does NOT
    register either timer tool.

    The default-off behavior matches the inner stack
    (``AgentDef.timers`` defaults to False there too) — agents
    that don't declare ``timers: true`` get the same minimal tool
    surface they did pre-step-10. A regression that flipped the
    default to True would surprise existing agents.
    """
    spec = AgentSpec(spec_version=1)  # timers defaults to False
    manager = ToolManager(spec=spec)
    names = manager.get_tool_names()
    assert SysTimerSetTool.name() not in names
    assert SysTimerCancelTool.name() not in names


def test_timers_true_registers_both_tools_and_schemas() -> None:
    """
    With ``timers=True`` the manager registers both tools and
    surfaces them in ``get_tool_schemas``.

    The schema visibility is what the LLM sees in its function
    list — a regression where registration succeeded but the
    schema didn't appear would manifest as the LLM holding a
    tool name it can't actually use (the runtime would still
    dispatch on call_tool, but the LLM never knew to call it).
    """
    spec = AgentSpec(spec_version=1, timers=True)
    manager = ToolManager(spec=spec)
    names = manager.get_tool_names()
    assert SysTimerSetTool.name() in names
    assert SysTimerCancelTool.name() in names
    schema_names = {s["function"]["name"] for s in manager.get_tool_schemas()}
    assert SysTimerSetTool.name() in schema_names
    assert SysTimerCancelTool.name() in schema_names


def test_timers_independent_of_async_enabled() -> None:
    """
    The ``timers`` and ``async_enabled`` flags are independent — a
    spec with ``timers=True, async_enabled=False`` registers the
    timer tools but NOT the async-inbox tools.

    Step 10 is described as building on the async-inbox machinery,
    but the LLM's tool surface is gated separately. An agent that
    wants timers without sys_call_async / sys_read_inbox /
    sys_cancel_async should be able to declare that combination
    explicitly.
    """
    spec = AgentSpec(spec_version=1, timers=True, async_enabled=False)
    manager = ToolManager(spec=spec)
    names = manager.get_tool_names()
    assert SysTimerSetTool.name() in names
    assert SysTimerCancelTool.name() in names
    # async-inbox tools are NOT registered.
    assert "sys_call_async" not in names
    assert "sys_read_inbox" not in names
    assert "sys_cancel_async" not in names