How to Test an AI Copilot / In-App Assistant

"""In-memory application state store standing in for a real SaaS backend.

The copilot harness needs something that behaves like your product's data layer
without the cost of a database: no I/O, no network, no fixtures to tear down, so
every test in this repo is fast and deterministic.

The shape here is a support ticket queue, because that is the worked example in
the blog post: a support lead asks the in-app assistant to "reassign every open
ticket from Jenna to Marcus, she's out sick today", and the only honest way to
know whether that worked is to look at the tickets afterwards.
"""

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import Dict, List, Optional

# Roles used throughout the harness. Keep this list tiny on purpose: permission
# scoping is only meaningful if you can read the whole matrix at a glance.
ROLE_AGENT = "agent"
ROLE_BILLING_ADMIN = "billing_admin"

STATUS_OPEN = "open"
STATUS_CLOSED = "closed"


class TicketNotFoundError(KeyError):
    """Raised when an action references a ticket id that does not exist."""


class UserNotFoundError(KeyError):
    """Raised when an action references a user id that does not exist."""


class RefundAmountError(ValueError):
    """Raised when a refund is non-positive or exceeds what is refundable."""


@dataclass
class Ticket:
    """One row in the support queue."""

    id: str
    subject: str
    customer: str
    owner: str
    status: str = STATUS_OPEN
    category: str = "general"
    amount: float = 0.0
    refunded_amount: float = 0.0

    @property
    def refundable_amount(self) -> float:
        return round(self.amount - self.refunded_amount, 2)


@dataclass
class User:
    """A person driving the product (and therefore the copilot)."""

    id: str
    name: str
    role: str


@dataclass
class AppStateStore:
    """The only thing in this repo that owns mutable application state.

    ``snapshot()`` is the workhorse for tests: it returns a deep-copied,
    plain-data view of everything that matters, so a test can diff before and
    after an action, or assert that a rejected action changed literally nothing.
    """

    tickets: Dict[str, Ticket] = field(default_factory=dict)
    users: Dict[str, User] = field(default_factory=dict)
    balances: Dict[str, float] = field(default_factory=dict)
    audit_log: List[str] = field(default_factory=list)

    # ---- lookups -------------------------------------------------------

    def get_ticket(self, ticket_id: str) -> Ticket:
        try:
            return self.tickets[ticket_id]
        except KeyError:
            raise TicketNotFoundError(f"no such ticket: {ticket_id!r}") from None

    def get_user(self, user_id: str) -> User:
        try:
            return self.users[user_id]
        except KeyError:
            raise UserNotFoundError(f"no such user: {user_id!r}") from None

    def filter_queue(
        self,
        owner: Optional[str] = None,
        status: Optional[str] = None,
        category: Optional[str] = None,
    ) -> List[Ticket]:
        """Read-only query over the queue. Never mutates anything."""
        matches = [
            ticket
            for ticket in self.tickets.values()
            if (owner is None or ticket.owner == owner)
            and (status is None or ticket.status == status)
            and (category is None or ticket.category == category)
        ]
        return sorted(matches, key=lambda ticket: ticket.id)

    # ---- mutations -----------------------------------------------------

    def reassign_ticket(self, ticket_id: str, new_owner: str) -> Ticket:
        """Move a single ticket to a new owner."""
        ticket = self.get_ticket(ticket_id)
        self.get_user(new_owner)  # reassigning to a ghost is a bug, not a feature
        previous_owner = ticket.owner
        ticket.owner = new_owner
        self.audit_log.append(
            f"reassign_ticket {ticket_id}: {previous_owner} -> {new_owner}"
        )
        return ticket

    def issue_refund(self, ticket_id: str, amount: float, requested_by: str) -> Ticket:
        """Refund money against a ticket.

        This is a second, independent permission check. The executor in
        ``copilot_app/actions.py`` already refuses to dispatch a refund for a
        user without the ``billing_admin`` role, but the store refuses too:
        money-moving operations should not be one missing ``if`` away from
        happening by accident.
        """
        requester = self.get_user(requested_by)
        if requester.role != ROLE_BILLING_ADMIN:
            raise PermissionError(
                f"user {requested_by!r} has role {requester.role!r}; "
                f"refunds require {ROLE_BILLING_ADMIN!r}"
            )

        ticket = self.get_ticket(ticket_id)
        amount = round(float(amount), 2)
        if amount <= 0:
            raise RefundAmountError(f"refund amount must be positive, got {amount}")
        if amount > ticket.refundable_amount:
            raise RefundAmountError(
                f"refund of {amount} exceeds refundable {ticket.refundable_amount} "
                f"on ticket {ticket_id}"
            )

        ticket.refunded_amount = round(ticket.refunded_amount + amount, 2)
        self.balances[ticket.customer] = round(
            self.balances.get(ticket.customer, 0.0) + amount, 2
        )
        self.audit_log.append(
            f"issue_refund {ticket_id}: {amount} to {ticket.customer} by {requested_by}"
        )
        return ticket

    # ---- snapshots -----------------------------------------------------

    def snapshot(self) -> dict:
        """A deep-copied, comparable view of current state.

        The audit log is deliberately excluded. Two snapshots should compare
        equal when the *data* is identical, so that a test can say "this
        rejected action changed nothing" without the log defeating it, and so
        that undo can restore data without rewriting history.
        """
        return {
            "tickets": {
                ticket_id: copy.deepcopy(vars(ticket))
                for ticket_id, ticket in sorted(self.tickets.items())
            },
            "users": {
                user_id: copy.deepcopy(vars(user))
                for user_id, user in sorted(self.users.items())
            },
            "balances": dict(sorted(self.balances.items())),
        }

    def restore(self, snapshot: dict) -> None:
        """Reset data state to a previously captured snapshot.

        Used by ``undo_last_action``. This is a rollback, not an action: the
        executor stays the only place a *proposed action* touches state.
        """
        restored = copy.deepcopy(snapshot)
        self.tickets = {
            ticket_id: Ticket(**payload)
            for ticket_id, payload in restored["tickets"].items()
        }
        self.users = {
            user_id: User(**payload) for user_id, payload in restored["users"].items()
        }
        self.balances = dict(restored["balances"])
        self.audit_log.append("restore: state reverted to snapshot")


def seed_store() -> AppStateStore:
    """Build the fixture used by every test in this repo.

    Five open tickets owned by ``jenna`` (the batch the copilot is asked to
    reassign), plus one closed ticket she owns, which must be left alone. That
    closed ticket is not padding: it is how the correctness test proves the
    copilot's ``status: open`` parameter was actually honored.
    """
    users = [
        User(id="marcus", name="Marcus Hale", role=ROLE_AGENT),
        User(id="jenna", name="Jenna Ruiz", role=ROLE_AGENT),
        User(id="priya", name="Priya Nair", role=ROLE_BILLING_ADMIN),
    ]
    tickets = [
        Ticket(
            id="T-1040",
            subject="Old invoice question",
            customer="acme",
            owner="jenna",
            status=STATUS_CLOSED,
            category="billing",
            amount=0.0,
        ),
        Ticket(
            id="T-1041",
            subject="Card declined on renewal",
            customer="acme",
            owner="jenna",
            status=STATUS_OPEN,
            category="failed_payment",
            amount=49.00,
        ),
        Ticket(
            id="T-1042",
            subject="Duplicate charge last Friday",
            customer="bolt",
            owner="jenna",
            status=STATUS_OPEN,
            category="billing",
            amount=12.50,
        ),
        Ticket(
            id="T-1043",
            subject="Cannot reset password",
            customer="cove",
            owner="jenna",
            status=STATUS_OPEN,
            category="general",
            amount=0.0,
        ),
        Ticket(
            id="T-1044",
            subject="Payment retry keeps failing",
            customer="dune",
            owner="jenna",
            status=STATUS_OPEN,
            category="failed_payment",
            amount=49.00,
        ),
        Ticket(
            id="T-1045",
            subject="Charged twice for seats",
            customer="echo",
            owner="jenna",
            status=STATUS_OPEN,
            category="billing",
            amount=25.00,
        ),
    ]
    return AppStateStore(
        tickets={ticket.id: ticket for ticket in tickets},
        users={user.id: user for user in users},
        balances={ticket.customer: 0.0 for ticket in tickets},
    )

"""The copilot client: turns a plain-language intent into a structured decision.

The wrapper's entire job is to force a decision out of the model. Not prose. An
action name and a set of parameters, validated against a small registry of
actions your application actually supports.

That constraint is what makes the rest of this repo testable. If free text is
the contract between the model and your application, every test you write has to
parse English to find out what the copilot decided, and every such test is one
rephrasing away from a false pass.

The provider call is injected (``llm_call``), so nothing here is tied to a
specific SDK. Pass a fake for tests, pass a real HTTP call for the nightly
regression run, pass your own vendor's client in production.
"""

from __future__ import annotations

import json
from collections.abc import Mapping
from typing import Any, Callable, Dict, Iterable, Union

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

from copilot_app.store import User

# The only actions this application recognizes. Anything else the model invents
# is rejected at the boundary rather than dispatched hopefully.
ACTIONS: tuple[str, ...] = ("reassign_tickets", "issue_refund", "filter_queue")

# A provider call takes a prompt and returns either a JSON string or an
# already-parsed mapping. Both are accepted so you can plug in a raw HTTP client
# or an SDK that parses for you.
LlmCall = Callable[[str], Union[str, Mapping[str, Any]]]

PROMPT_TEMPLATE = """You are the action-planning layer of an in-app assistant for a
support ticket queue. Choose exactly one action and return JSON only.

Allowed actions and parameters:
- reassign_tickets: {{"from": <owner id or null>, "to": <owner id>, "status": <"open"|"closed", optional>}}
- issue_refund: {{"ticket_id": <ticket id>, "amount": <number>}}
- filter_queue: {{"owner": <owner id, optional>, "status": <"open"|"closed", optional>, "category": <"failed_payment"|"billing"|"general", optional>}}

Known user ids: marcus, jenna, priya. Ticket ids look like T-1042.
Omit optional parameters entirely when the request does not mention them.

Respond with a single JSON object shaped like:
{{"action": "<one of {actions}>", "params": {{...}}, "rationale": "<one short sentence>", "confidence": <0.0-1.0>}}

REQUESTING_USER: {user_id} (role: {user_role})
INTENT: {intent}
"""


class CopilotProtocolError(ValueError):
    """The model returned something that is not a valid ProposedAction.

    Raised for malformed JSON, missing fields, an out-of-range confidence, or an
    action name outside the registry. A protocol error is a test failure, not
    something to recover from by guessing what the model meant.
    """


class ProposedAction(BaseModel):
    """Layer 2 of the three layers: the copilot's structured decision.

    ``rationale`` exists so a human can read why, and so a regression run can
    compare it loosely. It is never what a correctness test asserts on.
    """

    model_config = ConfigDict(extra="forbid")

    action: str
    params: Dict[str, Any] = Field(default_factory=dict)
    rationale: str = ""
    confidence: float = Field(default=0.0, ge=0.0, le=1.0)

    @field_validator("action")
    @classmethod
    def _action_must_be_registered(cls, value: str) -> str:
        if value not in ACTIONS:
            raise ValueError(
                f"unrecognized action {value!r}; allowed actions are {list(ACTIONS)}"
            )
        return value


class CopilotClient:
    """Wraps any model provider behind ``propose()``."""

    def __init__(
        self,
        llm_call: LlmCall,
        actions: Iterable[str] = ACTIONS,
        prompt_template: str = PROMPT_TEMPLATE,
    ) -> None:
        self._llm_call = llm_call
        self._actions = tuple(actions)
        self._prompt_template = prompt_template

    @property
    def actions(self) -> tuple[str, ...]:
        return self._actions

    def build_prompt(self, intent: str, user: User) -> str:
        return self._prompt_template.format(
            actions="|".join(self._actions),
            user_id=user.id,
            user_role=user.role,
            intent=intent,
        )

    def propose(self, intent: str, user: User) -> ProposedAction:
        """Ask the copilot what to do. Returns a decision, never prose."""
        raw = self._llm_call(self.build_prompt(intent, user))

        if isinstance(raw, (bytes, bytearray)):
            raw = raw.decode("utf-8")
        if isinstance(raw, str):
            try:
                raw = json.loads(_strip_code_fence(raw))
            except json.JSONDecodeError as exc:
                raise CopilotProtocolError(
                    f"copilot returned text that is not JSON: {raw[:400]!r}"
                ) from exc
        if not isinstance(raw, Mapping):
            raise CopilotProtocolError(
                f"copilot returned {type(raw).__name__}, expected a JSON object"
            )

        try:
            return ProposedAction.model_validate(dict(raw))
        except ValidationError as exc:
            raise CopilotProtocolError(
                f"copilot returned an invalid proposed action: {exc}"
            ) from exc


def _strip_code_fence(text: str) -> str:
    """Tolerate ```json fences, which most providers emit sooner or later."""
    stripped = text.strip()
    if not stripped.startswith("```"):
        return stripped
    lines = [line for line in stripped.splitlines() if not line.startswith("```")]
    return "\n".join(lines).strip()

"""The executor: the only place a proposed action is allowed to touch app state.

Two things make this file the load-bearing one in the harness.

First, permission scoping happens *before* dispatch. A copilot must never
execute an action the current user could not perform through the normal UI, and
a permission bypass is a different risk class from a merely wrong suggestion:
it ships silently even when every other test passes. So the check runs first and
raises before any handler is reached. Nothing is written, then rolled back.
Nothing is written at all.

Second, every successful execution returns an ``EffectDiff``: the before
snapshot, the after snapshot, and what ran. That diff is layer 3, the executed
effect, and it is the thing tests assert on instead of the copilot's wording.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Mapping

from copilot_app.copilot_client import ProposedAction
from copilot_app.store import AppStateStore, User

# The permission matrix, small enough to read in one glance and therefore small
# enough to review. An agent moves tickets around and filters the queue. Only a
# billing admin moves money.
PERMISSIONS: Mapping[str, frozenset] = {
    "agent": frozenset({"reassign_tickets", "filter_queue"}),
    "billing_admin": frozenset({"issue_refund"}),
}


class PermissionDeniedError(PermissionError):
    """The user's role does not include the proposed action.

    Subclasses the builtin ``PermissionError`` so callers that already handle
    permission failures generically keep working.
    """


class ActionDispatchError(ValueError):
    """The action is recognized and permitted but its parameters are unusable."""


@dataclass(frozen=True)
class EffectDiff:
    """What actually changed in the application, as plain comparable data."""

    action: str
    params: Dict[str, Any] = field(default_factory=dict)
    before: Dict[str, Any] = field(default_factory=dict)
    after: Dict[str, Any] = field(default_factory=dict)
    result: Any = None

    @property
    def changed(self) -> bool:
        return self.before != self.after

    def changed_ticket_ids(self) -> List[str]:
        """Ticket ids whose stored fields differ between before and after."""
        before_tickets = self.before.get("tickets", {})
        after_tickets = self.after.get("tickets", {})
        return sorted(
            ticket_id
            for ticket_id in set(before_tickets) | set(after_tickets)
            if before_tickets.get(ticket_id) != after_tickets.get(ticket_id)
        )


def permitted_actions(user: User) -> frozenset:
    return PERMISSIONS.get(user.role, frozenset())


def is_permitted(user: User, action: str) -> bool:
    return action in permitted_actions(user)


def execute_action(
    store: AppStateStore, action: ProposedAction, user: User
) -> EffectDiff:
    """Apply a proposed action to the store on behalf of ``user``.

    Raises ``PermissionDeniedError`` without touching the store when the action
    falls outside the user's own permissions, no matter how confident the
    copilot was when it proposed it.
    """
    before = store.snapshot()

    if not is_permitted(user, action.action):
        raise PermissionDeniedError(
            f"user {user.id!r} (role {user.role!r}) may not perform "
            f"{action.action!r}; permitted: {sorted(permitted_actions(user))}"
        )

    handler = _HANDLERS.get(action.action)
    if handler is None:  # pragma: no cover - registry and ACTIONS stay in sync
        raise ActionDispatchError(f"no handler registered for {action.action!r}")

    result = handler(store, dict(action.params), user)
    after = store.snapshot()

    return EffectDiff(
        action=action.action,
        params=dict(action.params),
        before=before,
        after=after,
        result=result,
    )


def _handle_reassign_tickets(
    store: AppStateStore, params: Dict[str, Any], user: User
) -> List[str]:
    new_owner = params.get("to")
    if not new_owner:
        raise ActionDispatchError("reassign_tickets requires a 'to' parameter")

    targets = store.filter_queue(owner=params.get("from"), status=params.get("status"))
    return [store.reassign_ticket(ticket.id, new_owner).id for ticket in targets]


def _handle_issue_refund(
    store: AppStateStore, params: Dict[str, Any], user: User
) -> str:
    ticket_id = params.get("ticket_id")
    amount = params.get("amount")
    if not ticket_id or amount is None:
        raise ActionDispatchError(
            "issue_refund requires 'ticket_id' and 'amount' parameters"
        )
    ticket = store.issue_refund(ticket_id, float(amount), requested_by=user.id)
    return ticket.id


def _handle_filter_queue(
    store: AppStateStore, params: Dict[str, Any], user: User
) -> List[str]:
    matches = store.filter_queue(
        owner=params.get("owner"),
        status=params.get("status"),
        category=params.get("category"),
    )
    return [ticket.id for ticket in matches]


_HANDLERS: Mapping[str, Callable[[AppStateStore, Dict[str, Any], User], Any]] = {
    "reassign_tickets": _handle_reassign_tickets,
    "issue_refund": _handle_issue_refund,
    "filter_queue": _handle_filter_queue,
}

"""Suggestion-to-action correctness: assert on what the app did.

The most valuable copilot test is boring by design. Give it an intent, let it
propose an action, execute that action against a realistic copy of the
application, and assert on the resulting state. Not on the copilot's sentence.
On the row that changed.

No network. The provider is the deterministic fake in
``copilot_app/fake_provider.py``, so these tests run in milliseconds and fail for
one reason only: the harness stopped doing the right thing to the store.
"""

from __future__ import annotations

import pytest

from copilot_app.actions import execute_action
from copilot_app.copilot_client import CopilotClient
from copilot_app.fake_provider import fake_llm_call
from copilot_app.store import seed_store

INTENT = "reassign every open ticket from Jenna to Marcus, she's out sick today"


@pytest.fixture
def store():
    return seed_store()


@pytest.fixture
def copilot():
    # The provider call is injected, so the same client class runs against a fake
    # here and against a real model in the nightly regression suite.
    return CopilotClient(llm_call=fake_llm_call)


def test_copilot_returns_a_structured_action_not_prose(store, copilot):
    proposed = copilot.propose(INTENT, store.users["marcus"])

    assert proposed.action == "reassign_tickets"
    assert proposed.params == {"from": "jenna", "to": "marcus", "status": "open"}


def test_executing_the_proposal_reassigns_every_open_jenna_ticket(store, copilot):
    marcus = store.users["marcus"]
    open_jenna_ticket_ids = [
        ticket.id for ticket in store.filter_queue(owner="jenna", status="open")
    ]
    assert len(open_jenna_ticket_ids) == 5, "fixture drift: expected 5 open tickets"

    proposed = copilot.propose(INTENT, marcus)
    diff = execute_action(store, proposed, marcus)

    # This is the assertion the copilot's cheerful summary can never make for
    # you. It reads the store, after the action ran, one ticket at a time.
    for ticket_id in open_jenna_ticket_ids:
        ticket = store.tickets[ticket_id]
        assert ticket.owner == "marcus"

    assert diff.changed
    assert diff.changed_ticket_ids() == sorted(open_jenna_ticket_ids)
    assert sorted(diff.result) == sorted(open_jenna_ticket_ids)


def test_the_closed_ticket_is_left_alone(store, copilot):
    marcus = store.users["marcus"]
    assert store.tickets["T-1040"].status == "closed"

    execute_action(store, copilot.propose(INTENT, marcus), marcus)

    # The copilot said "every open ticket". A run that also grabs the closed one
    # produces a plausible summary and the wrong queue.
    assert store.tickets["T-1040"].owner == "jenna"
    assert len(store.filter_queue(owner="marcus")) == 5
    assert len(store.filter_queue(owner="jenna")) == 1


def test_correctness_does_not_depend_on_the_rationale_text(store, copilot):
    marcus = store.users["marcus"]
    proposed = copilot.propose(INTENT, marcus)

    # Replace the copilot's explanation with something useless. The executed
    # effect, and therefore this suite, is unaffected. A prose assertion would
    # have broken here while the application behaved identically.
    rewritten = proposed.model_copy(update={"rationale": "Sure! All done. 🎉"})
    execute_action(store, rewritten, marcus)

    assert all(
        ticket.owner == "marcus"
        for ticket in store.filter_queue(status="open")
    )


def test_no_money_moved_and_no_ticket_was_refunded(store, copilot):
    marcus = store.users["marcus"]
    execute_action(store, copilot.propose(INTENT, marcus), marcus)

    assert all(balance == 0.0 for balance in store.balances.values())
    assert all(ticket.refunded_amount == 0.0 for ticket in store.tickets.values())

"""Permission scoping: a copilot must not become a privilege escalation path.

A wrong suggestion is a correctness bug. An action the requesting user could not
have performed through the normal UI is a different risk class, because it ships
silently even when every other test passes.

The assertion that matters here is not "an error was raised". It is "the store
never moved". A guard that raises after writing the row has not protected
anything, and only a before/after snapshot comparison can tell the difference.
"""

from __future__ import annotations

import pytest

from copilot_app.actions import PermissionDeniedError, execute_action, permitted_actions
from copilot_app.copilot_client import CopilotClient
from copilot_app.fake_provider import fake_llm_call
from copilot_app.store import seed_store

REFUND_INTENT = "issue a 49.00 refund on ticket T-1044, the customer was double charged"


@pytest.fixture
def store():
    return seed_store()


@pytest.fixture
def copilot():
    return CopilotClient(llm_call=fake_llm_call)


def test_the_copilot_happily_proposes_a_refund_to_an_agent(store, copilot):
    # The model is not the guardrail. It has no idea what this user may do, and
    # designing the test as though it did would hide the real bug.
    proposed = copilot.propose(REFUND_INTENT, store.users["marcus"])

    assert proposed.action == "issue_refund"
    assert proposed.params == {"ticket_id": "T-1044", "amount": 49.0}


def test_agent_cannot_execute_a_refund_and_the_store_never_moves(store, copilot):
    marcus = store.users["marcus"]
    assert "issue_refund" not in permitted_actions(marcus)

    proposed = copilot.propose(REFUND_INTENT, marcus)
    before = store.snapshot()

    with pytest.raises(PermissionDeniedError):
        execute_action(store, proposed, marcus)

    # The whole point. Not "an exception happened somewhere" but "no ticket, no
    # balance, and no owner field is different from before the attempt".
    assert store.snapshot() == before
    assert store.tickets["T-1044"].refunded_amount == 0.0
    assert store.balances["dune"] == 0.0


def test_billing_admin_executes_the_same_refund_successfully(store, copilot):
    priya = store.users["priya"]
    proposed = copilot.propose(REFUND_INTENT, priya)

    diff = execute_action(store, proposed, priya)

    assert diff.changed
    assert store.tickets["T-1044"].refunded_amount == 49.0
    assert store.tickets["T-1044"].refundable_amount == 0.0
    assert store.balances["dune"] == 49.0


def test_billing_admin_cannot_reassign_tickets(store, copilot):
    # Scoping cuts both ways. A billing admin holds the refund permission and
    # not the queue-management one.
    priya = store.users["priya"]
    proposed = copilot.propose(
        "reassign every open ticket from jenna to marcus", priya
    )
    before = store.snapshot()

    with pytest.raises(PermissionDeniedError):
        execute_action(store, proposed, priya)

    assert store.snapshot() == before


def test_agent_keeps_the_permissions_it_does_have(store, copilot):
    marcus = store.users["marcus"]
    proposed = copilot.propose(
        "reassign every open ticket from jenna to marcus", marcus
    )

    execute_action(store, proposed, marcus)

    assert len(store.filter_queue(owner="marcus", status="open")) == 5


def test_the_store_refuses_a_direct_refund_by_an_agent_too(store):
    # Defense in depth: even if a future caller bypasses the executor, the
    # money-moving method checks the requester's role itself.
    with pytest.raises(PermissionError):
        store.issue_refund("T-1044", 49.0, requested_by="marcus")

    assert store.tickets["T-1044"].refunded_amount == 0.0

"""The frozen scenario suite: checked-in data, not generated at runtime.

This file is the contract between your product and whatever model is behind the
copilot today. Each record pairs an intent with the exact action and parameters
it must produce. It is deliberately plain Python data so a support lead or a PM
can read every expected outcome without opening a test file.

Two rules for editing it:

1. Changing an expectation is a product decision, so it belongs in a pull
   request with a reason. Never edit a scenario to make a failing nightly run go
   green.
2. A scenario that no longer matches how the product should behave is worse than
   no test at all, because it looks like coverage while checking last quarter's
   assumptions. Delete it or fix it.

``expected_rationale`` is compared loosely (semantic similarity). Only
``expected_action`` and ``expected_params`` are matched exactly.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, Tuple


@dataclass(frozen=True)
class Scenario:
    """One frozen intent-to-action expectation."""

    id: str
    intent: str
    user_id: str
    role: str
    expected_action: str
    expected_params: Dict[str, Any] = field(default_factory=dict)
    expected_rationale: str = ""
    # True when the copilot is expected to propose this action and the executor
    # is expected to refuse it. Proposing a refund is not a bug; executing one
    # without the billing_admin role is.
    expect_permission_denied: bool = False
    # Part of the fast pull-request smoke set.
    smoke: bool = False
    notes: str = ""


FROZEN_SCENARIOS: Tuple[Scenario, ...] = (
    Scenario(
        id="reassign-open-tickets-from-jenna",
        intent="reassign every open ticket from Jenna to Marcus, she's out sick today",
        user_id="marcus",
        role="agent",
        expected_action="reassign_tickets",
        expected_params={"from": "jenna", "to": "marcus", "status": "open"},
        expected_rationale="Reassigning every open ticket owned by jenna to marcus.",
        smoke=True,
        notes="The worked example from the blog post. Must not touch closed tickets.",
    ),
    Scenario(
        id="reassign-same-request-different-words",
        intent="hand jenna's open tickets to marcus",
        user_id="marcus",
        role="agent",
        expected_action="reassign_tickets",
        expected_params={"from": "jenna", "to": "marcus", "status": "open"},
        expected_rationale="Reassigning every open ticket owned by jenna to marcus.",
        smoke=True,
        notes="Paraphrase of the scenario above. Same decision, different wording.",
    ),
    Scenario(
        id="refund-as-billing-admin",
        intent="issue a 49.00 refund on ticket T-1044, the customer was double charged",
        user_id="priya",
        role="billing_admin",
        expected_action="issue_refund",
        expected_params={"ticket_id": "T-1044", "amount": 49.0},
        expected_rationale="Refunding 49.00 against ticket T-1044 as requested.",
        smoke=True,
        notes="Full refund of the ticket amount by the one role allowed to move money.",
    ),
    Scenario(
        id="refund-as-agent-is-denied",
        intent="issue a 49.00 refund on ticket T-1044, the customer was double charged",
        user_id="marcus",
        role="agent",
        expected_action="issue_refund",
        expected_params={"ticket_id": "T-1044", "amount": 49.0},
        expected_rationale="Refunding 49.00 against ticket T-1044 as requested.",
        expect_permission_denied=True,
        smoke=True,
        notes=(
            "Identical intent to the scenario above, different role. The copilot is "
            "expected to propose the refund; the executor is expected to refuse it "
            "without touching the store."
        ),
    ),
    Scenario(
        id="refund-partial-amount",
        intent="refund 12.50 on ticket T-1045 for the duplicate charge",
        user_id="priya",
        role="billing_admin",
        expected_action="issue_refund",
        expected_params={"ticket_id": "T-1045", "amount": 12.5},
        expected_rationale="Refunding 12.50 against ticket T-1045 as requested.",
        notes="Partial refund: T-1045 is a 25.00 ticket.",
    ),
    Scenario(
        id="filter-open-failed-payments",
        intent="filter this queue to open failed payments",
        user_id="marcus",
        role="agent",
        expected_action="filter_queue",
        expected_params={"status": "open", "category": "failed_payment"},
        expected_rationale=(
            "Filtering the ticket queue to category=failed_payment, status=open."
        ),
        smoke=True,
        notes="Read-only action. Must not mutate the store.",
    ),
    Scenario(
        id="filter-jennas-open-queue",
        intent="show me everything still open in jenna's queue",
        user_id="marcus",
        role="agent",
        expected_action="filter_queue",
        expected_params={"owner": "jenna", "status": "open"},
        expected_rationale="Filtering the ticket queue to owner=jenna, status=open.",
        notes="Owner scoping comes from a possessive, not from an explicit field.",
    ),
)


def smoke_scenarios() -> Tuple[Scenario, ...]:
    """The subset the pull-request job runs, at a single repetition."""
    return tuple(scenario for scenario in FROZEN_SCENARIOS if scenario.smoke)


def scenario_by_id(scenario_id: str) -> Scenario:
    for scenario in FROZEN_SCENARIOS:
        if scenario.id == scenario_id:
            return scenario
    raise KeyError(f"no frozen scenario with id {scenario_id!r}")

"""Model-change regression: replay the frozen suite against the live copilot.

The failure mode that makes copilot testing different from testing a
conventional feature: the model provider ships an update on their schedule, not
yours. No pull request, no deploy, no changelog entry in your repo. The copilot
that used to propose "reassign to marcus" for a given intent now proposes
something else for the exact same words, because the decision boundary moved.

So this suite makes real provider calls, and it is built for a nightly cron
rather than for every commit:

* Each scenario runs ``REPETITIONS`` times (default 5) and needs
  ``PASS_THRESHOLD`` matches (default 4) to pass. One sample of a
  non-deterministic system is a coin flip, not a result.
* ``expected_action`` and ``expected_params`` are matched exactly. "Close
  enough" on which button got pressed is not close enough when the button
  reassigns five tickets or issues a refund.
* ``expected_rationale`` is compared with a loose similarity floor, because it is
  prose written for a human and rewording it is not a regression.
* A failure prints the frozen expectation next to what the copilot actually
  proposed, per attempt, so the diff is readable without a rerun.

With no provider credentials configured, the whole module skips instead of
failing, so a contributor can still run everything else in this repo.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

import pytest

from copilot_app.copilot_client import CopilotClient, CopilotProtocolError
from copilot_app.providers import build_live_llm_call, live_provider_configured
from copilot_app.similarity import DEFAULT_SIMILARITY_FLOOR, semantic_similarity
from copilot_app.store import seed_store
from scenarios.frozen_scenarios import FROZEN_SCENARIOS, Scenario

REPETITIONS = int(os.environ.get("COPILOT_REGRESSION_REPETITIONS", "5"))
PASS_THRESHOLD = int(os.environ.get("COPILOT_REGRESSION_PASS_THRESHOLD", "4"))
SIMILARITY_FLOOR = float(
    os.environ.get("COPILOT_RATIONALE_SIMILARITY_FLOOR", DEFAULT_SIMILARITY_FLOOR)
)

pytestmark = [
    pytest.mark.nightly,
    pytest.mark.skipif(
        not live_provider_configured(),
        reason=(
            "live provider not configured; set COPILOT_LLM_ENDPOINT and "
            "COPILOT_LLM_API_KEY to run the model-change regression suite"
        ),
    ),
]


@dataclass
class Attempt:
    """One replay of one scenario."""

    number: int
    action: Optional[str] = None
    params: Optional[Dict[str, Any]] = None
    rationale: str = ""
    rationale_similarity: float = 0.0
    error: Optional[str] = None
    mismatch_reasons: List[str] = field(default_factory=list)

    @property
    def matched(self) -> bool:
        return self.error is None and not self.mismatch_reasons


@pytest.fixture(scope="module")
def live_client() -> CopilotClient:
    return CopilotClient(llm_call=build_live_llm_call())


@pytest.mark.parametrize(
    "scenario", FROZEN_SCENARIOS, ids=[scenario.id for scenario in FROZEN_SCENARIOS]
)
def test_frozen_scenario_still_produces_the_same_action(
    scenario: Scenario, live_client: CopilotClient
) -> None:
    attempts = [
        _replay(scenario, live_client, attempt_number)
        for attempt_number in range(1, REPETITIONS + 1)
    ]
    matches = sum(1 for attempt in attempts if attempt.matched)

    if matches < PASS_THRESHOLD:
        pytest.fail(_report(scenario, attempts, matches), pytrace=False)


def _replay(scenario: Scenario, client: CopilotClient, attempt_number: int) -> Attempt:
    store = seed_store()  # a fresh store per attempt: no cross-attempt bleed
    user = store.get_user(scenario.user_id)
    assert user.role == scenario.role, (
        f"scenario {scenario.id!r} claims role {scenario.role!r} but user "
        f"{scenario.user_id!r} has role {user.role!r}"
    )

    attempt = Attempt(number=attempt_number)
    try:
        proposed = client.propose(scenario.intent, user)
    except CopilotProtocolError as exc:
        attempt.error = f"protocol error: {exc}"
        return attempt
    except Exception as exc:  # provider transport failures count as attempts
        attempt.error = f"{type(exc).__name__}: {exc}"
        return attempt

    attempt.action = proposed.action
    attempt.params = dict(proposed.params)
    attempt.rationale = proposed.rationale
    attempt.rationale_similarity = semantic_similarity(
        proposed.rationale, scenario.expected_rationale
    )

    # Exact on the decision.
    if proposed.action != scenario.expected_action:
        attempt.mismatch_reasons.append("action")
    if dict(proposed.params) != dict(scenario.expected_params):
        attempt.mismatch_reasons.append("params")
    # Loose on the prose.
    if attempt.rationale_similarity < SIMILARITY_FLOOR:
        attempt.mismatch_reasons.append("rationale")

    return attempt


def _report(scenario: Scenario, attempts: List[Attempt], matches: int) -> str:
    lines = [
        f"frozen scenario {scenario.id!r} regressed: "
        f"{matches}/{len(attempts)} matching runs, threshold {PASS_THRESHOLD}",
        "",
        f"intent          : {scenario.intent}",
        f"requesting user : {scenario.user_id} (role {scenario.role})",
        "",
        "FROZEN EXPECTATION",
        f"  action    : {scenario.expected_action}",
        f"  params    : {json.dumps(scenario.expected_params, sort_keys=True)}",
        f"  rationale : {scenario.expected_rationale}",
        f"              (compared loosely, similarity floor {SIMILARITY_FLOOR})",
        "",
        "WHAT THE COPILOT ACTUALLY PROPOSED",
    ]
    for attempt in attempts:
        verdict = "match" if attempt.matched else "MISMATCH"
        lines.append(f"  attempt {attempt.number} [{verdict}]")
        if attempt.error:
            lines.append(f"    error     : {attempt.error}")
            continue
        lines.append(f"    action    : {attempt.action}")
        lines.append(f"    params    : {json.dumps(attempt.params, sort_keys=True)}")
        lines.append(
            f"    rationale : {attempt.rationale!r} "
            f"(similarity {attempt.rationale_similarity})"
        )
        if attempt.mismatch_reasons:
            lines.append(f"    differs on: {', '.join(attempt.mismatch_reasons)}")
    if scenario.notes:
        lines += ["", f"note: {scenario.notes}"]
    return "\n".join(lines)

"""The full assistant flow, behaviorally: propose, confirm, execute, render, undo.

Everything else in this repo tests the copilot's decision in isolation. This
drives the sequence a real user actually walks through, and asserts on the two
boundaries that isolated tests never reach.

Confirm: a copilot that executes silently is riskier than one that proposes and
waits, because a wrong proposal caught at confirmation costs a click and a wrong
action executed unconfirmed costs a support ticket, or a customer's money. So
this test proves nothing runs before ``confirmed=True``.

Undo: the reversal gets its own assertion, against both the store and the queue
view, because "we call restore" and "the user sees the old queue again" are two
different claims.
"""

from __future__ import annotations

import pytest

from copilot_app.copilot_client import CopilotClient
from copilot_app.fake_provider import fake_llm_call
from copilot_app.session import (
    AssistantSession,
    ConfirmationRequiredError,
    NothingToUndoError,
)
from copilot_app.store import seed_store
from copilot_app.ui import render_ticket_queue

INTENT = "reassign every open ticket from Jenna to Marcus, she's out sick today"


@pytest.fixture
def store():
    return seed_store()


@pytest.fixture
def session(store):
    return AssistantSession(
        store=store,
        client=CopilotClient(llm_call=fake_llm_call),
        user=store.users["marcus"],
    )


def test_full_flow_from_intent_through_confirmation_to_undo(store, session):
    baseline = store.snapshot()
    opening_view = render_ticket_queue(store)
    assert opening_view.owner_counts == {"jenna": 6}

    # 1-2. The user opens the assistant and types an intent.
    proposed = session.submit_intent(INTENT)
    assert proposed.action == "reassign_tickets"
    assert proposed.params == {"from": "jenna", "to": "marcus", "status": "open"}

    # 3. A proposal is not an execution. Submitting the intent must not have
    #    changed a single field.
    assert store.snapshot() == baseline

    # 4. Confirm is the risk boundary. Without it, nothing runs.
    with pytest.raises(ConfirmationRequiredError):
        session.execute()
    assert store.snapshot() == baseline
    assert session.pending_action is proposed

    # 5. Confirmed. Now the app mutates state.
    diff = session.execute(confirmed=True)
    assert diff.changed
    assert len(diff.changed_ticket_ids()) == 5
    for ticket in store.filter_queue(status="open"):
        assert ticket.owner == "marcus"

    # 6. The UI reflects the change. Same data, projected the way a user reads it.
    view_after = render_ticket_queue(store)
    assert view_after.owner_counts == {"marcus": 5, "jenna": 1}
    assert view_after.row("T-1041")["assigned_to"] == "marcus"
    assert view_after.row("T-1041")["assigned_to_name"] == "Marcus Hale"
    assert view_after.row("T-1040")["assigned_to"] == "jenna"

    # 7. Undo reverts the change, in the store and in what the queue renders.
    assert session.can_undo
    undone = session.undo_last_action()
    assert undone.action == "reassign_tickets"
    assert store.snapshot() == baseline

    view_restored = render_ticket_queue(store)
    assert view_restored.owner_counts == {"jenna": 6}
    assert view_restored.owner_counts.get("marcus", 0) == 0
    assert view_restored.rows == opening_view.rows


def test_the_pending_proposal_is_cleared_once_executed(store, session):
    session.submit_intent(INTENT)
    session.execute(confirmed=True)

    assert session.pending_action is None
    assert len(session.executed) == 1


def test_a_truthy_value_is_not_a_confirmation(store, session):
    # ``confirmed`` is checked with ``is True``. A stray "yes" from a form field
    # is not a confirmation, and a coerced truthy value is exactly how an
    # unconfirmed action reaches production.
    baseline = store.snapshot()
    session.submit_intent(INTENT)

    with pytest.raises(ConfirmationRequiredError):
        session.execute(confirmed="yes")

    assert store.snapshot() == baseline


def test_undo_with_nothing_executed_raises(store, session):
    session.submit_intent(INTENT)

    with pytest.raises(NothingToUndoError):
        session.undo_last_action()


def test_undo_is_recorded_in_the_audit_log_rather_than_erasing_it(store, session):
    session.submit_intent(INTENT)
    session.execute(confirmed=True)
    session.undo_last_action()

    # Data state is reverted; the trail of what happened is not.
    assert any("reassign_ticket" in entry for entry in store.audit_log)
    assert store.audit_log[-1].startswith("restore:")

name: copilot-tests

# Two jobs, two very different costs.
#
# pr-smoke gates every pull request. It is deterministic, offline, and finishes
# in seconds, because nobody's pull request should sit behind a dozen real model
# calls.
#
# nightly-full makes real provider calls and replays the whole frozen suite at
# full repetition. It exists to catch the thing a pull request cannot: a
# provider-side model update that changed which action the copilot picks, with no
# commit anywhere in this repository.

on:
  pull_request:
  schedule:
    # 03:17 UTC daily. An odd minute so it is not queued alongside every other
    # cron job in the world.
    - cron: "17 3 * * *"

permissions:
  contents: read

jobs:
  pr-smoke:
    name: PR smoke (offline, deterministic)
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: pip

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Suggestion-to-action correctness, permissions, and the full flow
        run: |
          pytest -v \
            tests/test_suggestion_to_action.py \
            tests/test_permission_scoping.py \
            tests/test_e2e_assistant_flow.py

      - name: Frozen scenario smoke set (single repetition, fake provider)
        run: pytest -v -m smoke tests/test_frozen_scenarios_smoke.py

  nightly-full:
    name: Nightly full regression (real provider)
    if: github.event_name == 'schedule'
    runs-on: ubuntu-latest
    timeout-minutes: 30
    env:
      # Vendor-neutral: point these at any OpenAI-compatible endpoint, or swap
      # copilot_app/providers.py for your own provider call.
      COPILOT_LLM_ENDPOINT: ${{ secrets.COPILOT_LLM_ENDPOINT }}
      COPILOT_LLM_API_KEY: ${{ secrets.COPILOT_LLM_API_KEY }}
      COPILOT_LLM_MODEL: ${{ vars.COPILOT_LLM_MODEL }}
      # The non-determinism gate: repeat each scenario and require most runs to
      # match the frozen expectation.
      COPILOT_REGRESSION_REPETITIONS: "5"
      COPILOT_REGRESSION_PASS_THRESHOLD: "4"
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: pip

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Fail fast if the provider is not configured
        # Without this, the suite would skip and the job would go green, which is
        # the worst possible outcome for a nightly regression gate.
        run: |
          if [ -z "$COPILOT_LLM_API_KEY" ] || [ -z "$COPILOT_LLM_ENDPOINT" ]; then
            echo "::error::COPILOT_LLM_ENDPOINT and COPILOT_LLM_API_KEY must be set for the nightly run"
            exit 1
          fi

      - name: Replay every frozen scenario against the live copilot
        run: pytest -v -m nightly tests/test_model_change_regression.py