"""Owner approval for security-sensitive capabilities the coding agent declares.

Anonymous access, any delete, email/public-form primitives and out-of-plan packs are held for a
workspace admin; everything else provisions untouched. The verdict is decided fresh on every gate
call (the provisioning job's result is cached by manifest), is keyed by what is granted so it
survives retries, and a refusal sends the agent back through the existing correction path."""

from __future__ import annotations

import copy
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import uuid4

import pytest
from src.core.dependencies import require_project_codegen_admin
from src.core.exceptions import NotFoundError
from src.domain.codegen import capability_approval as ca
from src.domain.codegen.provision_gate import request_provisioning
from src.domain.codegen.work_branch import apply_report, bounded_correction, new_state, resume_plan

BASE = "a" * 40


def field(key="name"):
    return {"key": key, "label": key.title(), "type": "text"}


def entry(name="contact_submission", *, requirements, actors=None, fields=None, **extra):
    actors = actors or sorted({r["actor"] for r in requirements})
    return {"name": name, "actors": actors, "requirements": requirements,
            "fields": fields or [field("name"), field("phone")], **extra}


def req(actor, action="create", resource="contact_submission"):
    return {"actor": actor, "action": action, "resource": resource}


PUBLIC_FORM = entry(requirements=[req("anonymous"), req("admin", "read")])
ADMIN_ONLY = entry("internal_note", requirements=[req("admin", "create", "internal_note"),
                                                  req("admin", "read", "internal_note")])


# ---- what counts as sensitive ---------------------------------------------------------------------


def test_a_public_form_is_held_and_described_in_plain_words():
    [item] = ca.classify([PUBLIC_FORM], [], None)
    assert item.name == "contact_submission" and item.kind == "compose"
    assert "Anyone on the internet can create contact_submission records" in item.summary
    assert "Stores: name, phone" in item.summary
    assert {"actor": "anonymous", "action": "create", "resource": "contact_submission"} in item.grants


@pytest.mark.parametrize("actor", ["member", "admin", "super_admin"])
def test_ordinary_access_for_signed_in_actors_needs_nobody(actor):
    assert ca.classify([entry(requirements=[req(actor, "create"), req(actor, "read"),
                                             req(actor, "update")])], [], None) == []


def test_admin_only_data_is_not_held():
    assert ca.classify([ADMIN_ONLY], [], None) == []


@pytest.mark.parametrize("action", ["read", "update", "delete"])
def test_any_anonymous_access_is_held(action):
    assert len(ca.classify([entry(requirements=[req("anonymous", action)])], [], None)) == 1


def test_any_delete_is_held_even_for_an_admin():
    [item] = ca.classify([entry(requirements=[req("admin", "delete")])], [], None)
    assert "Admins can permanently delete contact_submission records" in item.summary


@pytest.mark.parametrize("primitive", ["email.send", "form.public_create"])
def test_the_email_and_public_form_primitives_are_held(primitive):
    [item] = ca.classify([entry(requirements=[req("admin", "read")], requires=[primitive])], [], None)
    assert item.grants == [{"primitive": primitive}]


def test_an_entry_is_judged_on_its_repaired_canonical_form():
    # unnormalized name and actor casing: repair makes it contact_submission / anonymous
    raw = entry("Contact Submission", requirements=[
        {"actor": " Anonymous ", "action": "CREATE", "resource": "Contact Submission"}])
    [item] = ca.classify([raw], [], None)
    assert item.name == "contact_submission"


def test_an_entry_that_is_still_invalid_after_repair_is_skipped_because_it_grants_nothing():
    assert ca.classify([{"name": "x", "requirements": "not a list"}], [], None) == []


def test_a_pack_outside_the_plan_is_held_only_when_there_is_a_plan_to_judge_against():
    installs = [{"pack_slug": "appointment-booking", "config": {}}, {"pack_slug": "chat", "config": {}}]
    [item] = ca.classify([], installs, ["appointment-booking"])
    assert item.kind == "pack_install" and item.name == "chat"
    assert "not part of this project's plan" in item.summary
    assert ca.classify([], installs, None) == []  # no plan, nothing to call an extra
    assert ca.classify([], installs, ["appointment-booking", "chat"]) == []


# ---- the fingerprint is the grant --------------------------------------------------------------


def test_the_fingerprint_ignores_cosmetics_but_changes_when_the_grant_changes():
    base = ca.classify([PUBLIC_FORM], [], None)[0].fingerprint
    cosmetic = copy.deepcopy(PUBLIC_FORM)
    cosmetic["fields"].append(field("email"))
    cosmetic["description"] = "A contact form"
    assert ca.classify([cosmetic], [], None)[0].fingerprint == base  # more fields, same grant
    wider = entry(requirements=[req("anonymous"), req("anonymous", "read"), req("admin", "read")])
    assert ca.classify([wider], [], None)[0].fingerprint != base  # public READ is a new question
    renamed = entry("lead", requirements=[req("anonymous", "create", "lead")])
    assert ca.classify([renamed], [], None)[0].fingerprint != base


# ---- decisions -----------------------------------------------------------------------------------


class MemoryStore:
    def __init__(self):
        self.rows = {}
        self.creates = 0

    async def get_many(self, project_id, fingerprints):
        return [r for r in self.rows.values() if r.project_id == project_id and r.fingerprint in fingerprints]

    async def create_pending(self, project_id, run_id, items):
        self.creates += 1
        for item in items:
            self.rows.setdefault((project_id, item.fingerprint), SimpleNamespace(
                id=uuid4(), project_id=project_id, fingerprint=item.fingerprint, kind=item.kind,
                name=item.name, summary=item.summary, grants=item.grants, status="pending",
                requested_run_id=run_id, decided_by=None, decided_at=None))

    async def get(self, project_id, approval_id):
        return next((r for r in self.rows.values()
                     if r.project_id == project_id and r.id == approval_id), None)

    async def set_status(self, row, status, user_id):
        row.status, row.decided_by, row.decided_at = status, user_id, datetime.now(UTC)

    async def list(self, project_id):
        return [r for r in self.rows.values() if r.project_id == project_id]


@pytest.mark.asyncio
async def test_a_new_grant_is_recorded_once_as_pending_however_often_the_gate_asks():
    store, project, run = MemoryStore(), uuid4(), uuid4()
    items = ca.classify([PUBLIC_FORM], [], None)
    first = await ca.evaluate(store, project, run, items)
    again = await ca.evaluate(store, project, run, items)
    assert first.created and len(first.pending) == 1
    assert not again.created and len(again.pending) == 1 and len(store.rows) == 1


@pytest.mark.asyncio
async def test_an_approval_carries_over_and_only_a_changed_grant_asks_again():
    store, project = MemoryStore(), uuid4()
    items = ca.classify([PUBLIC_FORM], [], None)
    await ca.evaluate(store, project, uuid4(), items)
    [row] = await store.list(project)
    await ca.decide(store, project, row.id, True, uuid4())
    later = await ca.evaluate(store, project, uuid4(), items)  # a different run, same grant
    assert later.clear and len(later.approved) == 1 and not later.created
    wider = ca.classify([entry(requirements=[req("anonymous"), req("anonymous", "read")])], [], None)
    changed = await ca.evaluate(store, project, uuid4(), wider)
    assert len(changed.pending) == 1 and not changed.clear  # the new grant is a new question
    assert (await store.list(project))[0].status == "approved"  # the old decision is untouched


@pytest.mark.asyncio
async def test_approvals_never_leak_across_projects():
    store, items = MemoryStore(), ca.classify([PUBLIC_FORM], [], None)
    a, b = uuid4(), uuid4()
    await ca.evaluate(store, a, uuid4(), items)
    await ca.decide(store, a, (await store.list(a))[0].id, True, uuid4())
    other = await ca.evaluate(store, b, uuid4(), items)
    assert len(other.pending) == 1  # project b was never asked, so it is not approved


@pytest.mark.asyncio
async def test_deciding_records_who_and_when_and_can_be_changed_later():
    store, project, admin = MemoryStore(), uuid4(), uuid4()
    await ca.evaluate(store, project, uuid4(), ca.classify([PUBLIC_FORM], [], None))
    [row] = await store.list(project)
    rejected = await ca.decide(store, project, row.id, False, admin)
    assert rejected.status == "rejected" and rejected.decided_by == admin and rejected.decided_at
    assert (await ca.decide(store, project, row.id, True, admin)).status == "approved"


@pytest.mark.asyncio
async def test_deciding_an_unknown_or_other_projects_approval_is_not_found():
    store, project = MemoryStore(), uuid4()
    await ca.evaluate(store, project, uuid4(), ca.classify([PUBLIC_FORM], [], None))
    [row] = await store.list(project)
    with pytest.raises(NotFoundError):
        await ca.decide(store, project, uuid4(), True, uuid4())
    with pytest.raises(NotFoundError):
        await ca.decide(store, uuid4(), row.id, True, uuid4())


# ---- the gate verdict ------------------------------------------------------------------------------


def make_run(**plan_context):
    return SimpleNamespace(id=uuid4(), project_id=uuid4(), status="running",
                           plan={"context": plan_context})


def repository():
    return SimpleNamespace(db=object(), commit=AsyncMock())


@pytest.fixture
def store(monkeypatch):
    memory = MemoryStore()
    monkeypatch.setattr(ca, "SqlApprovalStore", lambda db: memory)
    return memory


@pytest.mark.asyncio
async def test_an_unapproved_public_form_blocks_the_run_and_says_how_to_continue(store):
    repo, run = repository(), make_run()
    held = await ca.verdict(repo, run, [], [PUBLIC_FORM])
    assert held["status"] == "blocked"
    assert held["failure_class"] == "approval_required" and held["retryable"] is True
    assert "contact_submission" in held["message"] and "Anyone on the internet" in held["message"]
    assert "dashboard" in held["message"] and "retry" in held["message"]
    assert "correction" not in held  # nothing for the agent to change yet
    repo.commit.assert_awaited_once()  # the pending rows are durable before the runner is told


@pytest.mark.asyncio
async def test_asking_again_while_still_pending_does_not_write_again(store):
    repo, run = repository(), make_run()
    await ca.verdict(repo, run, [], [PUBLIC_FORM])
    await ca.verdict(repo, run, [], [PUBLIC_FORM])
    assert repo.commit.await_count == 1 and store.creates == 1


@pytest.mark.asyncio
async def test_once_approved_the_gate_lets_it_through(store):
    repo, run = repository(), make_run()
    await ca.verdict(repo, run, [], [PUBLIC_FORM])
    [row] = await store.list(run.project_id)
    await ca.decide(store, run.project_id, row.id, True, uuid4())
    assert await ca.verdict(repo, run, [], [PUBLIC_FORM]) is None


@pytest.mark.asyncio
async def test_a_rejection_sends_the_agent_back_through_the_correction_path(store):
    repo, run = repository(), make_run()
    await ca.verdict(repo, run, [], [PUBLIC_FORM])
    [row] = await store.list(run.project_id)
    await ca.decide(store, run.project_id, row.id, False, uuid4())
    refused = await ca.verdict(repo, run, [], [PUBLIC_FORM])
    assert refused["failure_class"] == "invalid_requirement" and refused["retryable"] is True
    kept = bounded_correction(refused["correction"])
    assert kept["entries"][0]["name"] == "contact_submission"
    error = kept["entries"][0]["errors"][0]
    assert error["code"] == "not_approved" and "did not approve" in error["message"]
    assert "Anyone on the internet" in error["message"]


@pytest.mark.asyncio
async def test_ordinary_declarations_never_touch_the_store(store):
    repo = repository()
    assert await ca.verdict(repo, make_run(), [], [ADMIN_ONLY]) is None
    assert store.creates == 0 and store.rows == {}
    repo.commit.assert_not_awaited()


@pytest.mark.asyncio
async def test_out_of_plan_packs_are_judged_against_the_generation_plan(store):
    plan = {"generation_plan": {"composition": {"feature_pack_ids": ["appointment-booking"]}}}
    held = await ca.verdict(repository(), make_run(**plan), [{"pack_slug": "chat", "config": {}}], [])
    assert held["failure_class"] == "approval_required" and "chat" in held["message"]
    assert await ca.verdict(repository(), make_run(),
                            [{"pack_slug": "chat", "config": {}}], []) is None  # no plan


# ---- what happens next -----------------------------------------------------------------------------


def blocked_state(failure_class, correction=None):
    return apply_report(new_state(uuid4(), "preview", BASE), status="blocked", checkpoint="agent_output",
                        checkpoint_sha="b" * 40, failure_class=failure_class, retryable=True,
                        correction=correction)


def test_a_retry_after_approval_skips_the_agent_and_only_redoes_validation():
    assert resume_plan(blocked_state("approval_required"), max_corrections=2) == {
        "skip_agent": True, "last_checkpoint": "agent_output"}


def test_a_retry_after_a_rejection_sends_the_agent_back_with_the_refusal():
    correction = {"entries": [{"name": "contact_submission", "errors": [
        {"path": "requirements", "code": "not_approved", "message": "The project owner did not approve"}]}]}
    plan = resume_plan(blocked_state("invalid_requirement", correction), max_corrections=2)
    assert plan["skip_agent"] is False and plan["correction"]["entries"][0]["name"] == "contact_submission"


# ---- the gate itself -----------------------------------------------------------------------------


class FakeRedis:
    def __init__(self):
        self.enqueued = []

    async def enqueue_job(self, *args, **kwargs):
        self.enqueued.append(args)

    async def aclose(self):
        return None


async def gate(monkeypatch, run, compositions):
    redis = FakeRedis()

    async def get_pool():
        return redis

    monkeypatch.setattr("src.workers.enqueue._get_redis_pool", get_pool)

    class Job:
        def __init__(self, *args):
            pass

        async def result_info(self):
            return None

    monkeypatch.setattr("src.domain.codegen.provision_gate.Job", Job)
    repo = SimpleNamespace(
        get_by_id=AsyncMock(return_value=run), get_project_discovery=AsyncMock(return_value=None),
        db=object(), commit=AsyncMock())
    return await request_provisioning(repo, run.id, {}, [], compositions), redis


@pytest.mark.asyncio
async def test_the_gate_holds_a_public_form_before_any_provisioning_job_is_queued(monkeypatch, store):
    run = make_run()
    result, redis = await gate(monkeypatch, run, [PUBLIC_FORM])
    assert result["failure_class"] == "approval_required"
    assert redis.enqueued == []  # nothing is provisioned in Cloud for an undecided grant


@pytest.mark.asyncio
async def test_after_approval_the_same_manifest_is_provisioned_and_not_served_a_stale_verdict(monkeypatch, store):
    run = make_run()
    await gate(monkeypatch, run, [PUBLIC_FORM])
    [row] = await store.list(run.project_id)
    await ca.decide(store, run.project_id, row.id, True, uuid4())
    result, redis = await gate(monkeypatch, run, [PUBLIC_FORM])
    assert result == {"status": "pending"} and len(redis.enqueued) == 1


@pytest.mark.asyncio
async def test_an_ordinary_manifest_goes_straight_to_provisioning(monkeypatch, store):
    result, redis = await gate(monkeypatch, make_run(), [ADMIN_ONLY])
    assert result == {"status": "pending"} and len(redis.enqueued) == 1


# ---- who may decide --------------------------------------------------------------------------------


def dependency_names(path_suffix):
    from src.domain.codegen.router import router

    route = next(r for r in router.routes if r.path.endswith(path_suffix))
    return [d.call for d in route.dependant.dependencies]


@pytest.mark.parametrize("suffix", ["/approve", "/reject"])
def test_only_workspace_admins_can_decide(suffix):
    # workspaces.manage: owners and admins. Not editors or developers, and never an AI connection
    # (agent access has its own principal and no route to this cookie-authorized endpoint).
    assert require_project_codegen_admin in dependency_names("/capability-approvals/{approval}" + suffix)


def test_members_can_list_but_the_list_is_not_a_decision_route():
    from src.domain.codegen.router import router

    listing = next(r for r in router.routes if r.path.endswith("/capability-approvals"))
    assert "GET" in listing.methods and "POST" not in listing.methods
    assert require_project_codegen_admin not in dependency_names("/capability-approvals")


# ---- the calling agent is told ---------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_build_says_the_build_is_waiting_for_the_owner_so_the_agent_does_not_just_retry():
    from tests.test_agent_tool_audit import _get_build_result

    state = blocked_state("approval_required")
    waiting = await _get_build_result(
        status="blocked", error_message="Waiting for approval: 1 security-sensitive capability ...",
        plan={"context": {"work_branch": state}})
    assert waiting["needs_owner_approval"] is True
    no_flag = await _get_build_result(
        status="blocked", error_message="Waiting for approval: 1 security-sensitive capability ...",
        plan={"context": {}})
    assert no_flag["needs_owner_approval"] is True  # works without the work branch too
    other = await _get_build_result(status="blocked", error_message="A capability is missing",
                                    plan={"context": {"work_branch": blocked_state("capability_gap")}})
    assert other["needs_owner_approval"] is False
    done = await _get_build_result(status="completed", error_message=None,
                                   plan={"context": {"work_branch": state}})
    assert done["needs_owner_approval"] is False