FabledGrimoire/tests/test_reply_shape_turn.py

"""The core reply shape rides every turn.

Every turn ends in a reply, and the operator's prompt is the last point before
it is written, so the per-turn retrieval carries the core shape ahead of
everything else: in full the first time a session sees it, as its one-line
reminder after that, and in full again once a compaction has swept the
ledger. What is pinned here is that round trip across the shell/Python seam —
the route's order and keys, and the hook's ledger.
"""
from __future__ import annotations

import inspect
import json
import os
import subprocess
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

from quart import Quart, g

from grimoire.services import reply_shapes
from tests.helpers import http_sink, need_tools

ROOT = Path(__file__).resolve().parents[1]
HOOK = ROOT / "plugin" / "hooks" / "grimoire_autoinject.sh"
SESSION_START = ROOT / "plugin" / "hooks" / "grimoire_session_context.sh"


# ── the route ────────────────────────────────────────────────────────────


async def _retrieve(query_string, turn_context="CORE SHAPE"):
    from grimoire.routes import plugin as routes

    turn = AsyncMock(return_value={"context": turn_context, "rule_ids": [173],
                                   "shape_forms": {"core": reply_shapes.FULL}})
    app = Quart(__name__)
    async with app.test_request_context("/api/plugin/retrieve", query_string=query_string):
        g.user = SimpleNamespace(id=7)
        with patch.object(routes.plugin_ctx_svc, "build_prompt_rule_hint",
                          AsyncMock(return_value={"context": "RULE LINE", "rule_ids": [11],
                                                  "shown_rule_ids": [11]})), \
             patch.object(routes.plugin_ctx_svc, "build_autoinject_hint",
                          AsyncMock(return_value={"context": "NOTES MENU", "note_ids": []})), \
             patch.object(routes.lesson_rules_svc, "co_surfaced", AsyncMock(return_value="")), \
             patch.object(routes.moment_delivery_svc, "deliver_for_turn", turn):
            resp = await inspect.unwrap(routes.autoinject_retrieve)()
            return await resp.get_json(), turn


async def test_the_turns_shape_leads_the_payload_and_its_key_comes_back():
    body, _turn = await _retrieve({"q": "fix the flaky thing"})
    ctx = body["context"]
    assert ctx.index("CORE SHAPE") < ctx.index("RULE LINE") < ctx.index("NOTES MENU")
    assert body["shape_keys"] == ["core"]
    # Both arms' fresh rules reach the shared ledger.
    assert body["rule_ids"] == [11, 173]


async def test_the_route_hands_the_ledger_and_the_rules_just_named_to_the_turn():
    _body, turn = await _retrieve({"q": "x", "shapes_seen": "core,bogus",
                                   "exclude_rule_ids": "9", "held_rule_ids": "4"})
    kw = turn.await_args.kwargs
    assert kw["seen"] == frozenset({"core"})
    # A rule the prompt arm just named is not quoted again by the turn.
    assert kw["exclude"] == frozenset({9, 11})
    assert kw["held"] == frozenset({4})


# ── the hook ─────────────────────────────────────────────────────────────


def _env(tmp_path, port):
    return {"PATH": os.environ["PATH"], "GRIMOIRE_URL": f"http://127.0.0.1:{port}",
            "GRIMOIRE_TOKEN": "t", "TMPDIR": str(tmp_path), "HOME": str(tmp_path)}


def _prompt(tmp_path, port, session="s-shape"):
    need_tools("bash", "curl", "awk")
    out = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({"session_id": session, "cwd": str(tmp_path),
                          "prompt": "please fix the importer"}),
        capture_output=True, text=True, env=_env(tmp_path, port), timeout=30,
    )
    assert out.returncode == 0, out.stderr
    return out.stdout


def _compact(tmp_path, session="s-shape"):
    out = subprocess.run(
        ["bash", str(SESSION_START)],
        input=json.dumps({"session_id": session, "source": "compact"}),
        capture_output=True, text=True,
        env={"PATH": os.environ["PATH"], "HOME": str(tmp_path), "TMPDIR": str(tmp_path)},
        timeout=60,
    )
    assert out.returncode == 0, out.stderr


REPLY = json.dumps({"context": "Reply shape · Every reply", "note_ids": [], "rule_ids": [],
                    "shape_keys": ["core"]}).encode()


def test_the_core_goes_whole_once_then_as_a_reminder_then_whole_after_a_compaction(tmp_path):
    with http_sink(by_path={"/api/plugin/retrieve": REPLY}) as (port, seen):
        out = _prompt(tmp_path, port)
        _prompt(tmp_path, port)
        _compact(tmp_path)
        _prompt(tmp_path, port)
    asks = [e for e in seen if e["_path"] == "/api/plugin/retrieve"]
    assert len(asks) == 3
    assert "shapes_seen" not in asks[0]
    assert asks[1]["shapes_seen"] == ["core"]
    # The compaction swept the ledger: the session no longer holds the shape.
    assert "shapes_seen" not in asks[2]
    assert "Every reply" in json.loads(out)["hookSpecificOutput"]["additionalContext"]


def test_the_hook_and_the_route_agree_on_the_ledgers_names():
    """A contract across the seam: a renamed field fails silently."""
    from grimoire.routes import plugin as routes

    hook = HOOK.read_text()
    assert "shapes_seen" in hook and "'.shape_keys'" in hook
    assert "grimoire_shapes_file" in hook and "grimoire_shapes_append" in hook
    route = inspect.getsource(routes.autoinject_retrieve)
    assert 'request.args.get("shapes_seen")' in route and '"shape_keys"' in route