plyngent/tests/test_agent/test_notices.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta, timezone

from plyngent.agent import (
    ChatAgent,
    Notice,
    aside_notice,
    format_moment,
    has_notice,
    resume_notice,
    skill_tools_notice,
    truncated_turn_notice,
)
from plyngent.agent.chat import incomplete_turn_user_text
from plyngent.config.models import DatabaseConfig
from plyngent.lmproto.openai_compatible.model import (
    AssistantChatMessage,
    AssistantFunctionTool,
    AssistantFunctionToolCall,
    DeveloperChatMessage,
    SystemChatMessage,
    ToolChatMessage,
    UserChatMessage,
)
from plyngent.memory import MemoryStore
from tests.test_agent.test_loop import ScriptedClient


def test_notice_message_carries_marker_and_kind() -> None:
    message = Notice(kind="resume", body="  hello there  ").to_message()
    assert isinstance(message, DeveloperChatMessage)
    assert message.content == "[plyngent notice: resume]\nhello there"


def test_format_moment_localizes_and_labels_missing() -> None:
    assert format_moment(None) == "unknown"
    aware = datetime(2026, 9, 14, 22, 9, tzinfo=UTC)
    assert format_moment(aware) == aware.astimezone().strftime("%Y-%m-%d %H:%M")
    # SQLite hands back naive datetimes for tz-aware columns; read them as UTC.
    assert format_moment(aware.replace(tzinfo=None)) == format_moment(aware)
    offset = datetime(2026, 9, 14, 22, 9, tzinfo=timezone(timedelta(hours=8)))
    assert format_moment(offset) == offset.astimezone().strftime("%Y-%m-%d %H:%M")


def test_resume_notice_states_what_is_gone_and_restored() -> None:
    notice = resume_notice(last_active=datetime(2026, 9, 14, 22, 9, tzinfo=UTC))
    assert notice.kind == "resume"
    assert "resumed in a new plyngent process" in notice.body
    assert "PTY sessions" in notice.body
    assert "grants were restored" in notice.body
    assert notice.summary.startswith("resumed in a new process")
    assert format_moment(datetime(2026, 9, 14, 22, 9, tzinfo=UTC)) in notice.summary


def test_aside_notice_matches_tool_scope() -> None:
    read = aside_notice(tools_mode="read")
    assert read.kind == "aside"
    assert "read-only" in read.body
    assert "not saved" in read.body
    assert "read tools" in read.summary

    assert "Tools are disabled" in aside_notice(tools_mode="no").body
    assert "Full tools are available" in aside_notice(tools_mode="full").body
    # Unknown mode degrades to the read-only wording rather than an empty notice.
    assert aside_notice(tools_mode="weird").body == read.body


def test_skill_tools_notice_points_at_the_tools_without_listing() -> None:
    notice = skill_tools_notice()
    assert notice.kind == "skills"
    # Instructions only — what exists is asked for, never injected.
    assert "not listed here" in notice.body
    assert "`skill_list`" in notice.body
    assert "`skill_search`" in notice.body
    assert "`skill_read NAME [file]`" in notice.body
    assert "data, not code" in notice.body
    assert "skill_list" in notice.summary


def test_has_notice_finds_a_persisted_row_by_kind() -> None:
    messages = [
        SystemChatMessage(content="sys"),
        skill_tools_notice().to_message(),
        resume_notice(last_active=None).to_message(),
    ]
    assert has_notice(messages, "skills") is True
    assert has_notice(messages, "resume") is True
    assert has_notice(messages, "aside") is False
    # A plain developer row (a directive checkpoint) is not a notice.
    assert has_notice([DeveloperChatMessage(content="checkpoint")], "skills") is False


async def test_push_notice_appends_and_persists() -> None:
    store = await MemoryStore.open(DatabaseConfig())
    try:
        session = await store.create_session(name="t")
        _ = await store.append_message(session.sid, SystemChatMessage(content="sys"))
        _ = await store.append_message(session.sid, UserChatMessage(content="hi"))
        agent = ChatAgent(ScriptedClient([]), model="m", memory=store, session_id=session.sid)
        await agent.load_history()

        message = await agent.push_notice(resume_notice(last_active=None))

        assert agent.messages[-1] is message
        assert agent.persist_from == len(agent.messages)  # never re-persisted by a batch
        loaded = await store.list_messages(session.sid)
        assert isinstance(loaded[-1], DeveloperChatMessage)
        assert loaded[-1].content.startswith("[plyngent notice: resume]")
    finally:
        await store.close()


def test_trailing_developer_message_keeps_incomplete_turn_detectable() -> None:
    """A pushed notice must not hide a turn that can still be retried."""
    notice = resume_notice(last_active=None).to_message()
    assert incomplete_turn_user_text([UserChatMessage(content="do it"), notice]) == "do it"
    batch = [
        UserChatMessage(content="do it"),
        AssistantChatMessage(
            content="",
            tool_calls=[
                AssistantFunctionToolCall(
                    id="1",
                    function=AssistantFunctionTool(name="ping", arguments="{}"),
                )
            ],
        ),
        ToolChatMessage(content="pong", tool_call_id="1"),
        notice,
    ]
    assert incomplete_turn_user_text(batch) == "do it"
    # A finished turn (trailing assistant) stays non-retryable past a notice.
    done = [UserChatMessage(content="do it"), AssistantChatMessage(content="ok"), notice]
    assert incomplete_turn_user_text(done) is None


def test_truncated_turn_notice_keeps_the_partial_answer_retryable() -> None:
    """A kept partial answer is unfinished work, so retry continues it."""
    tail = [
        UserChatMessage(content="do it"),
        AssistantChatMessage(content="partial answer"),
        truncated_turn_notice().to_message(),
    ]
    assert incomplete_turn_user_text(tail) == "do it"
    # The same tail without the truncation notice is a finished turn.
    assert incomplete_turn_user_text(tail[:2]) is None


def test_truncated_turn_notice_tells_the_model_where_it_stopped() -> None:
    notice = truncated_turn_notice()
    assert notice.kind == "truncated"
    assert notice.to_message().content.startswith("[plyngent notice: truncated]")
    assert "max_tokens" in notice.body
    # The dropped call and the kept answer are both spelled out.
    assert "discarded" in notice.body
    assert "kept in the transcript" in notice.body
    assert "max_tokens" in notice.summary