"""Contracts for the engine-owned structured terminal action."""

from __future__ import annotations

import json

import pytest

from openai4s.agent.actions import (
    CodeCell,
    FinalizeAction,
    NativeToolBatch,
    NativeToolCall,
    route_action,
)
from openai4s.agent.engine import AgentEngine
from openai4s.agent.events import (
    ActionRouted,
    OutcomeProduced,
    ReplyReceived,
    RunFinished,
)
from openai4s.agent.finalize import (
    execute_finalize_action,
    execution_evidence,
    finalize_response_schema,
    finalize_response_tool_spec,
    reconcile_completion_claims,
    validate_finalize_arguments,
    with_finalize_response,
)
from openai4s.agent.ledger import RuntimeActionLedger, restore_action_history
from openai4s.agent.models import EngineResult, ModelReply, RunState
from openai4s.agent.runtime import LocalActionExecutor
from openai4s.server.action_timeline import ActionTimelineService
from openai4s.server.agent_run import WebActionExecutor, WebEventSink
from openai4s.store import Store
from openai4s.tools import REGISTRY, ToolSpec


def _arguments(**overrides):
    value = {
        "summary": "The inspected evidence supports the requested conclusion.",
        "findings": ["The control and measured value agree."],
        "metrics": {"accuracy": 0.93},
        "artifacts": ["artifact-1", "prediction.csv"],
        "limitations": ["Only one dataset was available."],
        "next_steps": ["Validate on an independent dataset."],
        "completion_bullets": ["Completed the evidence review"],
    }
    value.update(overrides)
    return value


def _prose_arguments(**overrides):
    """A payload with no execution-shaped claims — no artifacts, no metrics,
    reflective bullets — which is what an honest conversational turn (zero
    cells, zero tools) is allowed to finalize with under ledger reconciliation.
    """
    value = {
        "summary": "The inspected evidence supports the requested conclusion.",
        "findings": ["The control and measured value agree."],
        "limitations": ["Only one dataset was available."],
        "next_steps": ["Validate on an independent dataset."],
        "completion_bullets": ["Explained the evidence review"],
    }
    value.update(overrides)
    return value


def _call(
    arguments=None,
    *,
    call_id="final-1",
    name="finalize_response",
    parse_error=None,
):
    raw = json.dumps(arguments, ensure_ascii=False) if arguments is not None else "{}"
    return NativeToolCall(
        id=call_id,
        wire_id=f"wire-{call_id}",
        name=name,
        ordinal=0,
        raw_arguments=raw,
        arguments=arguments,
        parse_error=parse_error,
        provider_meta={"provider": "test"},
    )


class _NeverKernel:
    generation = 0

    def execute(self, *args, **kwargs):
        raise AssertionError(
            f"structured finalization started a kernel: {args!r} {kwargs!r}"
        )


class _NeverDispatcher:
    last_output = None

    def __call__(self, *args, **kwargs):
        raise AssertionError(
            f"structured finalization dispatched a tool: {args!r} {kwargs!r}"
        )


def _local_executor():
    return LocalActionExecutor(
        _NeverKernel(),
        _NeverDispatcher(),
        lambda code, messages: None,
        lambda code: (_ for _ in ()).throw(
            AssertionError(f"structured finalization started R: {code}")
        ),
    )


def _web_executor(*, cancelled=lambda: False, plan_mode=False):
    sent = []
    events = WebEventSink(sent.append, "frame-1", [], lambda usage: None)

    def unexpected(*args, **kwargs):
        raise AssertionError(
            f"structured finalization ran Web work: {args!r} {kwargs!r}"
        )

    return WebActionExecutor(
        dispatcher=lambda: unexpected,
        apply_pending=unexpected,
        execute_cell=unexpected,
        events=events,
        prose_nudge="nudge",
        explore_nudge="explore",
        cancelled=cancelled,
        plan_mode=plan_mode,
    )


def test_finalize_spec_is_closed_host_strict_and_outside_control_registry():
    spec = finalize_response_tool_spec()

    assert isinstance(spec, ToolSpec)
    assert spec.name == "finalize_response"
    assert spec.strict is False
    assert spec.input_schema["additionalProperties"] is False
    assert set(spec.input_schema["required"]) == {"summary", "completion_bullets"}
    assert set(spec.input_schema["properties"]) == {
        "summary",
        "findings",
        "metrics",
        "artifacts",
        "limitations",
        "next_steps",
        "completion_bullets",
        "task_status",
        # Code-mode evidence: optional on the wire and closed on the schema, so
        # an analysis turn is unaffected while a reusable_pipeline /
        # codebase_change turn has somewhere honest to put its deliverable.
        "source_files",
        "entry_points",
        "architecture_summary",
        "test_evidence",
    }
    # There is deliberately no field for a test's OUTPUT text: pass/fail is
    # read off the stored stdout of the named cell, so "the tests passed"
    # cannot be a claim the payload carries.
    assert set(
        spec.input_schema["properties"]["test_evidence"]["items"]["properties"]
    ) == {"command", "producing_cell_id"}
    assert spec.input_schema["properties"]["task_status"]["enum"] == [
        "completed",
        "partial",
        "blocked",
        "failed",
    ]
    # Optional: the declaration is honesty support, never a new requirement.
    assert "task_status" not in spec.input_schema["required"]
    assert "sole tool call" in spec.description
    assert "finalize_response" not in {tool.name for tool in REGISTRY}

    spec.input_schema["properties"]["summary"]["maxLength"] = 1
    assert finalize_response_schema()["properties"]["summary"]["maxLength"] == 4_000

    catalogue = with_finalize_response((ToolSpec("lookup", "", {}),))
    assert [item.name for item in catalogue] == ["lookup", "finalize_response"]
    with pytest.raises(ValueError, match="engine-owned"):
        with_finalize_response(
            ({"type": "function", "function": {"name": "finalize_response"}},)
        )


def test_host_validation_rejects_missing_unknown_and_semantically_bad_fields():
    assert validate_finalize_arguments(_arguments()) is None
    assert "required property" in validate_finalize_arguments(
        {"completion_bullets": ["Completed the task"]}
    )
    assert "unknown property" in validate_finalize_arguments(
        _arguments(unverified_claim=True)
    )
    cardinality = validate_finalize_arguments(_arguments(completion_bullets=[]))
    assert "completion_bullets" in cardinality and ">= 1" in cardinality
    assert "past-tense" in validate_finalize_arguments(
        _arguments(completion_bullets=["Finish the task"])
    )


def test_finalize_accepts_a_declared_task_status_and_rejects_garbage():
    """D5: the closed schema takes an optional task_status enum, and the
    accepted value rides into the completion record's output for the
    delegation envelope's single-writer derivation to read."""
    assert validate_finalize_arguments(_arguments(task_status="partial")) is None
    assert validate_finalize_arguments(_arguments(task_status="blocked")) is None
    error = validate_finalize_arguments(_arguments(task_status="almost done"))
    assert error is not None and "task_status" in error
    error = validate_finalize_arguments(_arguments(task_status=True))
    assert error is not None and "task_status" in error

    outcome = execute_finalize_action(
        FinalizeAction(_call(_arguments(task_status="partial")))
    )
    assert outcome.completion is not None
    assert outcome.completion["output"]["task_status"] == "partial"

    refused = execute_finalize_action(
        FinalizeAction(_call(_arguments(task_status="done")))
    )
    assert refused.completion is None
    assert refused.history_messages[0]["is_error"] is True


def test_router_reserves_finalization_only_for_one_standalone_native_call():
    final = _call(_arguments())
    control = _call({"path": "."}, call_id="list-1", name="list_dir")

    assert route_action("ordinary prose") is None
    assert route_action("", (final,)) == FinalizeAction(final)

    mixed = route_action("", (control, final))
    assert isinstance(mixed, NativeToolBatch)
    assert mixed.calls == (control, final)

    duplicate = route_action("", (final, _call(_arguments(), call_id="final-2")))
    assert isinstance(duplicate, NativeToolBatch)
    assert route_action("```python\nprint(1)\n```", ()) == CodeCell(
        "python", "print(1)\n"
    )


def test_cli_executor_closes_provider_call_before_returning_completion_record():
    call = _call(_prose_arguments())
    outcome = _local_executor().execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), RunState([])
    )

    assert outcome.history_messages == (
        {
            "role": "tool",
            "tool_call_id": "final-1",
            "wire_id": "wire-final-1",
            "name": "finalize_response",
            "content": '{"status":"accepted","action":"finalize_response"}',
            "is_error": False,
        },
    )
    assert outcome.completion == {
        "output": {
            key: value
            for key, value in _prose_arguments().items()
            if key != "completion_bullets"
        },
        "completion_bullets": ["Explained the evidence review"],
    }
    assert outcome.stop_reason is None


def test_invalid_finalize_is_a_canonical_error_result_and_does_not_complete():
    call = _call({"summary": "Incomplete", "completion_bullets": []})
    outcome = execute_finalize_action(FinalizeAction(call))

    assert len(outcome.history_messages) == 1
    result = outcome.history_messages[0]
    assert result["role"] == "tool"
    assert result["tool_call_id"] == call.id
    assert result["is_error"] is True
    assert "completion_bullets" in result["content"]
    assert outcome.completion is None

    malformed = _call(None, call_id="bad-json", parse_error="invalid JSON")
    malformed_outcome = execute_finalize_action(FinalizeAction(malformed))
    assert malformed_outcome.history_messages[0]["tool_call_id"] == "bad-json"
    assert "invalid JSON" in malformed_outcome.history_messages[0]["content"]
    assert malformed_outcome.completion is None


def test_mixed_batch_treats_finalize_as_nonterminal_and_never_completes():
    control = _call({"path": "."}, call_id="list-1", name="list_dir")
    final = _call(_arguments())

    class Dispatcher:
        last_output = None

        def __call__(self, method, args):
            assert method == "list_dir" and args == [{"path": "."}]
            return {"entries": []}

    executor = LocalActionExecutor(
        _NeverKernel(),
        Dispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )
    outcome = executor.execute(
        NativeToolBatch((control, final)),
        ModelReply(tool_calls=(control, final)),
        RunState([]),
    )

    assert outcome.completion is None
    assert outcome.stop_reason is None
    assert [message["tool_call_id"] for message in outcome.history_messages] == [
        "list-1",
        "final-1",
    ]
    assert outcome.history_messages[-1]["is_error"] is True
    assert "unknown tool" in outcome.history_messages[-1]["content"]


def test_engine_records_assistant_then_tool_result_before_submitted_terminal():
    call = _call(_prose_arguments())
    reply = ModelReply(content="", tool_calls=(call,))

    class Model:
        def complete(self, messages, on_delta):
            del messages, on_delta
            return reply

    result = AgentEngine(Model(), _local_executor(), max_turns=1).run(
        [{"role": "user", "content": "Summarize the completed review."}]
    )

    assert result.stop_reason == "submitted"
    assert result.turns == 1
    assert [message["role"] for message in result.messages] == [
        "user",
        "assistant",
        "tool",
    ]
    assert result.messages[-1]["tool_call_id"] == call.id
    assert result.completion["output"]["summary"].startswith("The inspected")


def test_web_executor_accepts_finalize_without_dispatcher_kernel_or_pending_work():
    call = _call(_prose_arguments())
    outcome = _web_executor().execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), RunState([])
    )

    assert outcome.history_messages[0]["tool_call_id"] == call.id
    assert outcome.history_messages[0]["is_error"] is False
    assert outcome.completion is not None


def test_web_cancel_and_plan_close_finalize_call_without_completion():
    call = _call(_arguments())
    reply = ModelReply(tool_calls=(call,))

    cancelled = _web_executor(cancelled=lambda: True).execute(
        FinalizeAction(call), reply, RunState([])
    )
    assert cancelled.stop_reason == "cancelled"
    assert cancelled.history_messages[0]["is_error"] is True
    assert cancelled.completion is None

    planned = _web_executor(plan_mode=True).execute(
        FinalizeAction(call), reply, RunState([])
    )
    assert planned.stop_reason == "plan"
    assert planned.history_messages[0]["is_error"] is True
    assert planned.completion is None


def test_finalize_ledger_roundtrip_and_timeline_projection(tmp_path):
    store = Store(tmp_path / "openai4s.db")
    ledger = RuntimeActionLedger(store, "root-final", "turn-final")
    call = _call(_arguments())
    reply = ModelReply(tool_calls=(call,))
    action = FinalizeAction(call)
    outcome = execute_finalize_action(action)

    ledger.append_user("Finish the response")
    ledger.emit(ReplyReceived(reply, 0))
    ledger.emit(ActionRouted(action, 0))
    ledger.emit(OutcomeProduced(outcome, 0))
    ledger.emit(
        RunFinished(EngineResult((), outcome.completion, "submitted", 1, reply))
    )

    groups = store.list_action_groups("root-final")
    assert [group["kind"] for group in groups] == ["user", "finalize", "terminal"]
    assert groups[1]["events"][0]["resource_keys"] == ["agent:completion"]
    assert groups[1]["events"][1]["result"]["is_error"] is False

    history = restore_action_history(store, "root-final")
    assert [message["role"] for message in history] == ["user", "assistant", "tool"]
    assert history[-1]["tool_call_id"] == call.id

    timeline = ActionTimelineService(store).get("root-final")
    finalized = next(
        group for group in timeline["groups"] if group["kind"] == "finalize"
    )
    assert finalized["status"] == "completed"
    assert finalized["title"].startswith("The inspected evidence")
    store.close()


def test_reconciliation_rejects_execution_claims_without_ledger_evidence():
    zero = {"cells": 0, "tool_calls": 0}

    assert reconcile_completion_claims(_prose_arguments(), zero) is None
    error = reconcile_completion_claims(_arguments(), zero)
    assert "artifacts" in error and "metrics" in error and "ledger" in error
    flagged = reconcile_completion_claims(
        _prose_arguments(completion_bullets=["Computed the standard deviation"]),
        zero,
    )
    assert "Computed the standard deviation" in flagged
    # CJK bullets carry no tense morphology; the verb heuristic never applies.
    assert (
        reconcile_completion_claims(
            _prose_arguments(completion_bullets=["完成了证据审阅"]), zero
        )
        is None
    )
    # Any real execution satisfies the reconciliation for every claim shape.
    assert reconcile_completion_claims(_arguments(), {"cells": 1}) is None
    assert reconcile_completion_claims(_arguments(), {"tool_calls": 2}) is None
    # A corrupted evidence mapping degrades to zero counts, never a crash.
    assert "ledger" in reconcile_completion_claims(
        _arguments(), {"cells": "three", "tool_calls": -1}
    )


def test_zero_execution_finalize_with_claims_is_refused_but_repairable():
    call = _call(_arguments(completion_bullets=["Computed the standard deviation"]))

    for executor in (_local_executor(), _web_executor()):
        outcome = executor.execute(
            FinalizeAction(call), ModelReply(tool_calls=(call,)), RunState([])
        )
        result = outcome.history_messages[0]
        assert result["is_error"] is True
        assert "ledger" in result["content"]
        assert outcome.completion is None
        # A refused reconciliation is repairable, not terminal: the model can
        # run the claimed work (or restate the completion) on the next turn.
        assert outcome.stop_reason is None

    # Direct calls without an evidence ledger keep the legacy behaviour.
    legacy = execute_finalize_action(FinalizeAction(call))
    assert legacy.history_messages[0]["is_error"] is False
    assert legacy.completion is not None


def test_cli_finalize_accepts_execution_claims_after_a_real_cell_ran():
    class Kernel:
        generation = 0

        def execute(self, code, origin="agent"):
            assert origin == "agent"
            return {"stdout": "4\n", "stderr": "", "error": None}

    class Dispatcher:
        last_output = None

    executor = LocalActionExecutor(
        Kernel(),
        Dispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )
    state = RunState([])
    cell = CodeCell("python", "print(2 + 2)\n")
    executor.execute(cell, ModelReply(content="```python\nprint(2 + 2)\n```"), state)
    assert execution_evidence(state.metadata) == {"cells": 1, "tool_calls": 0}

    call = _call(_arguments(completion_bullets=["Computed the requested value"]))
    outcome = executor.execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), state
    )
    assert outcome.history_messages[0]["is_error"] is False
    assert outcome.completion is not None


def test_safety_refused_cell_never_counts_as_execution_evidence():
    executor = LocalActionExecutor(
        _NeverKernel(),
        _NeverDispatcher(),
        lambda code, messages: "cell refused by the safety gate",
        lambda code: {"error": "R must not start"},
    )
    state = RunState([])
    cell = CodeCell("python", "import os\n")
    refused = executor.execute(
        cell, ModelReply(content="```python\nimport os\n```"), state
    )
    assert "refused" in str(refused.observation)
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}

    call = _call(_arguments())
    outcome = executor.execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), state
    )
    assert outcome.history_messages[0]["is_error"] is True
    assert outcome.completion is None


def test_web_finalize_accepts_execution_claims_after_a_real_cell_ran():
    sent = []
    events = WebEventSink(sent.append, "frame-1", [], lambda usage: None)

    class Dispatcher:
        last_output = None

    executor = WebActionExecutor(
        dispatcher=lambda: Dispatcher(),
        apply_pending=lambda: None,
        execute_cell=lambda action: {"stdout": "ok\n", "stderr": "", "error": None},
        events=events,
        prose_nudge="nudge",
        explore_nudge="explore",
    )
    state = RunState([])
    cell = CodeCell("python", "print(1)\n")
    executor.execute(cell, ModelReply(content="```python\nprint(1)\n```"), state)
    assert execution_evidence(state.metadata) == {"cells": 1, "tool_calls": 0}

    call = _call(_arguments(completion_bullets=["Computed the requested value"]))
    outcome = executor.execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), state
    )
    assert outcome.history_messages[0]["is_error"] is False
    assert outcome.completion is not None


def test_web_refused_cell_result_never_counts_as_execution_evidence():
    """The Web sibling of the safety-gate contract.

    The gateway's ``execute_cell`` returns its full outcome dict whose
    ``executed`` bit is False for safety-refused / runtime-unavailable soft
    errors — result dicts byte-identical to real failures.  Counting them let
    a Web turn whose only cell was refused finalize with fabricated
    execution claims.
    """
    sent = []
    events = WebEventSink(sent.append, "frame-1", [], lambda usage: None)

    class Dispatcher:
        last_output = None

    executor = WebActionExecutor(
        dispatcher=lambda: Dispatcher(),
        apply_pending=lambda: None,
        execute_cell=lambda action: {
            "result": {"stdout": "", "stderr": "", "error": "refused"},
            "executed": False,
        },
        events=events,
        prose_nudge="nudge",
        explore_nudge="explore",
    )
    state = RunState([])
    cell = CodeCell("python", "import os\n")
    executor.execute(cell, ModelReply(content="```python\nimport os\n```"), state)
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}

    call = _call(_arguments(completion_bullets=["Computed the requested value"]))
    outcome = executor.execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), state
    )
    assert outcome.history_messages[0]["is_error"] is True
    assert outcome.completion is None


def test_refused_native_batch_never_counts_as_execution_evidence():
    """A batch whose every call was refused without dispatch is not evidence.

    Parse/validation/limit refusals are answered by ``execute_native_batch``
    itself — no tool runs.  Counting declarations let one malformed call back
    a fabricated finalize.
    """
    bad = NativeToolCall(
        id="t-1",
        wire_id="wire-t-1",
        name="read_file",
        ordinal=0,
        raw_arguments="{not json",
        arguments=None,
        parse_error="arguments are not valid JSON",
    )

    class NeverInvokedDispatcher:
        last_output = None

        def execute_host_call(self, *args, **kwargs):  # pragma: no cover
            raise AssertionError("a refused call must not reach the dispatcher")

    local = LocalActionExecutor(
        _NeverKernel(),
        NeverInvokedDispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )
    state = RunState([])
    local.execute(NativeToolBatch((bad,)), ModelReply(tool_calls=(bad,)), state)
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}

    call = _call(_arguments(completion_bullets=["Queried the database"]))
    refused = local.execute(FinalizeAction(call), ModelReply(tool_calls=(call,)), state)
    assert refused.history_messages[0]["is_error"] is True
    assert refused.completion is None


def test_cli_r_unavailable_soft_error_never_counts_as_execution_evidence():
    """The R runner's spawn-failure dict carries no "stdout": nothing ran."""
    executor = LocalActionExecutor(
        _NeverKernel(),
        _NeverDispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R kernel unavailable: Rscript not found"},
    )
    state = RunState([])
    cell = CodeCell("r", "summary(1:3)\n")
    outcome = executor.execute(
        cell, ModelReply(content="```r\nsummary(1:3)\n```"), state
    )
    assert "R kernel unavailable" in str(outcome.observation)
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}


def test_call_reaches_dispatcher_gates_evidence_counting():
    """Evidence counts only calls that would actually run a tool.

    A hallucinated tool name, or a known tool whose arguments would be
    refused, executes nothing — counting it let a refused call back a later
    execution-shaped finalize claim.
    """
    from openai4s.agent.control import call_reaches_dispatcher

    # A known tool with valid arguments will be dispatched → counts.
    assert call_reaches_dispatcher("read_text_file", None, {"path": "a.txt"}) is True
    # The native path validated arguments before invoke, so a bare known name
    # (no arguments passed) counts too.
    assert call_reaches_dispatcher("list_dir", None) is True
    # An unknown / hallucinated name is refused before the dispatcher.
    assert call_reaches_dispatcher("run_python_cell", None, {"code": "x"}) is False
    # A known tool with a type-invalid argument is refused before the dispatcher
    # (this is the legacy path's gap: it has no prior validate step).
    assert call_reaches_dispatcher("read_text_file", None, {"path": 123}) is False
    # Degenerate names never count.
    assert call_reaches_dispatcher(None) is False
    assert call_reaches_dispatcher("") is False


def test_unknown_native_tool_name_never_counts_as_execution_evidence():
    """A hallucinated native tool name slips past the batch validator (which
    only knows how to reject *known* tools' bad input) but is refused before
    the dispatcher — it must not become finalize-time evidence."""
    unknown = NativeToolCall(
        id="t-1",
        wire_id="wire-1",
        name="run_python_cell",
        ordinal=0,
        raw_arguments='{"code": "print(1)"}',
        arguments={"code": "print(1)"},
        parse_error=None,
    )

    class NeverInvokedDispatcher:
        last_output = None

        def __call__(self, *a, **k):  # pragma: no cover - unknown tool early-exits
            raise AssertionError("an unknown tool must not reach the dispatcher")

    local = LocalActionExecutor(
        _NeverKernel(),
        NeverInvokedDispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )
    state = RunState([])
    local.execute(NativeToolBatch((unknown,)), ModelReply(tool_calls=(unknown,)), state)
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}

    call = _call(_arguments(completion_bullets=["Ran the analysis"]))
    refused = local.execute(FinalizeAction(call), ModelReply(tool_calls=(call,)), state)
    assert refused.history_messages[0]["is_error"] is True
    assert refused.completion is None


def test_legacy_unknown_or_invalid_tool_calls_never_count_as_evidence():
    """The text-parsed path has no prior validate gate, so it must itself
    refuse to count an unknown tool or a known tool with invalid arguments."""
    local = LocalActionExecutor(
        _NeverKernel(),
        _NeverDispatcher(),
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )
    for content in (
        '```tool\n{"name": "run_python_cell", "arguments": {"code": "x"}}\n```',
        '```tool\n{"name": "read_text_file", "arguments": {"path": 123}}\n```',
    ):
        state = RunState([])
        local.execute(None, ModelReply(content=content), state)
        assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 0}


# --- a dispatched-but-refused call is not execution evidence ---------------
#
# ``call_reaches_dispatcher`` only proves a call *would* be dispatched. The
# dispatcher's own permission gate, the tool's static precheck, a soft-fail
# ``{"error": ...}`` result and a raised dispatch all come after it, and none of
# them executed the work a later bullet may claim. Evidence therefore counts a
# native call only once its dispatch reports ``ok``.

_DENIAL = "Permission denied: approval required but no interactive channel is attached"
_ZERO_EVIDENCE = {"cells": 0, "tool_calls": 0}


def _native(name, arguments, *, call_id="t-1"):
    return NativeToolCall(
        id=call_id,
        wire_id=f"wire-{call_id}",
        name=name,
        ordinal=0,
        raw_arguments=json.dumps(arguments),
        arguments=arguments,
        parse_error=None,
    )


class _ScriptedHost:
    """A HostDispatcher double answering each method with a fixed result."""

    last_output = None

    def __init__(self, result):
        self.result = result
        self.methods: list[str] = []

    def __call__(self, method, args):
        self.methods.append(method)
        if isinstance(self.result, BaseException):
            raise self.result
        return self.result


def _local_with(dispatcher):
    return LocalActionExecutor(
        _NeverKernel(),
        dispatcher,
        lambda code, messages: None,
        lambda code: {"error": "R must not start"},
    )


def _web_with(dispatcher):
    sent = []
    return WebActionExecutor(
        dispatcher=lambda: dispatcher,
        apply_pending=lambda: None,
        execute_cell=lambda action: (_ for _ in ()).throw(
            AssertionError("a native-tool turn ran a cell")
        ),
        events=WebEventSink(sent.append, "frame-1", [], lambda usage: None),
        prose_nudge="nudge",
        explore_nudge="explore",
    )


def _assert_claim_refused(executor, state):
    call = _call(_prose_arguments(completion_bullets=["Wrote out.csv"]))
    outcome = executor.execute(
        FinalizeAction(call), ModelReply(tool_calls=(call,)), state
    )
    assert outcome.history_messages[0]["is_error"] is True
    assert "ledger" in outcome.history_messages[0]["content"]
    assert outcome.completion is None


@pytest.mark.parametrize("make_executor", [_local_with, _web_with], ids=["cli", "web"])
def test_permission_denied_native_call_never_counts_as_execution_evidence(
    make_executor,
):
    host = _ScriptedHost({"error": _DENIAL})
    executor = make_executor(host)
    denied = _native("write_file", {"path": "out.csv", "content": "x,y\n"})
    state = RunState([])

    outcome = executor.execute(
        NativeToolBatch((denied,)), ModelReply(tool_calls=(denied,)), state
    )

    # The call really reached the dispatcher, whose gate refused it.
    assert host.methods == ["write_file"]
    assert outcome.history_messages[0]["is_error"] is True
    assert "Permission denied" in outcome.history_messages[0]["content"]
    assert execution_evidence(state.metadata) == _ZERO_EVIDENCE
    _assert_claim_refused(executor, state)


@pytest.mark.parametrize("make_executor", [_local_with, _web_with], ids=["cli", "web"])
def test_raised_native_dispatch_never_counts_as_execution_evidence(make_executor):
    host = _ScriptedHost(RuntimeError("dispatcher exploded"))
    executor = make_executor(host)
    call = _native("list_dir", {"path": "."})
    state = RunState([])

    outcome = executor.execute(
        NativeToolBatch((call,)), ModelReply(tool_calls=(call,)), state
    )

    assert host.methods == ["list_dir"]
    assert outcome.history_messages[0]["is_error"] is True
    assert execution_evidence(state.metadata) == _ZERO_EVIDENCE


@pytest.mark.parametrize("make_executor", [_local_with, _web_with], ids=["cli", "web"])
def test_precheck_refused_native_call_never_counts_as_execution_evidence(
    make_executor,
):
    host = _ScriptedHost({"ok": True})
    executor = make_executor(host)
    noop = _native(
        "edit_file", {"path": "a.txt", "old_string": "same", "new_string": "same"}
    )
    state = RunState([])

    outcome = executor.execute(
        NativeToolBatch((noop,)), ModelReply(tool_calls=(noop,)), state
    )

    assert host.methods == []
    assert "no-op edit" in outcome.history_messages[0]["content"]
    assert execution_evidence(state.metadata) == _ZERO_EVIDENCE
    _assert_claim_refused(executor, state)


@pytest.mark.parametrize("make_executor", [_local_with, _web_with], ids=["cli", "web"])
def test_only_the_succeeded_calls_of_a_mixed_native_batch_count(make_executor):
    class PerMethodHost(_ScriptedHost):
        def __call__(self, method, args):
            self.methods.append(method)
            if method == "write_file":
                return {"error": _DENIAL}
            return {"entries": ["a.txt"]}

    host = PerMethodHost(None)
    executor = make_executor(host)
    batch = NativeToolBatch(
        (
            _native("list_dir", {"path": "."}, call_id="t-1"),
            _native("write_file", {"path": "b.txt", "content": "b"}, call_id="t-2"),
        )
    )
    state = RunState([])

    executor.execute(batch, ModelReply(tool_calls=batch.calls), state)

    assert host.methods == ["list_dir", "write_file"]
    # The successful read is still evidence: counting stays conservative.
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 1}


@pytest.mark.parametrize("make_executor", [_local_with, _web_with], ids=["cli", "web"])
def test_legacy_permission_denied_tool_call_never_counts_as_evidence(make_executor):
    host = _ScriptedHost({"error": _DENIAL})
    executor = make_executor(host)
    content = (
        '```tool\n{"name": "write_file", "arguments": '
        '{"path": "out.csv", "content": "x"}}\n```'
    )
    state = RunState([])

    outcome = executor.execute(None, ModelReply(content=content), state)

    assert host.methods == ["write_file"]
    assert "Permission denied" in str(outcome.observation)
    assert execution_evidence(state.metadata) == _ZERO_EVIDENCE
    _assert_claim_refused(executor, state)

    succeeded = _ScriptedHost({"entries": []})
    control = make_executor(succeeded)
    state = RunState([])
    control.execute(
        None,
        ModelReply(
            content='```tool\n{"name": "list_dir", "arguments": {"path": "."}}\n```'
        ),
        state,
    )
    assert succeeded.methods == ["list_dir"]
    assert execution_evidence(state.metadata) == {"cells": 0, "tool_calls": 1}


def test_cli_run_denied_native_tool_cannot_back_a_wrote_claim(monkeypatch, tmp_path):
    """End to end through the CLI composition under the unattended deny posture.

    A model that issues one approval-gated call, is denied, and then finalizes
    with an execution claim must be refused; the honest restatement is what
    the run submits.
    """
    import openai4s.agent.loop as loop_mod
    from openai4s.agent import Agent

    monkeypatch.setenv("OPENAI4S_UNATTENDED_APPROVAL", "deny")
    scripted = [
        (
            # The live pattern: a no-op dynamic tool right before finalizing.
            "define_dynamic_tool",
            {
                "name": "noop",
                "description": "noop",
                "input_schema": {"type": "object"},
                "output_schema": {"type": "object"},
                "implementation": "def execute(args):\n    return {}\n",
            },
        ),
        (
            "finalize_response",
            {
                "summary": "Done.",
                "completion_bullets": ["Wrote the classification cell"],
            },
        ),
        (
            "finalize_response",
            {
                "summary": "The file could not be written.",
                "completion_bullets": ["Explained the permission limitation"],
            },
        ),
    ]
    observations: list[str] = []

    def chat(messages, cfg, **kwargs):
        del cfg, kwargs
        observations.extend(
            str(message.get("content") or "")
            for message in messages
            if message.get("role") == "tool"
        )
        name, arguments = scripted.pop(0)
        call = {
            "id": f"call-{len(scripted)}",
            "wire_id": f"wire-{len(scripted)}",
            "name": name,
            "ordinal": 0,
            "raw_arguments": json.dumps(arguments),
            "arguments": arguments,
            "parse_error": None,
        }
        return {
            "content": "",
            "tool_calls": [call],
            "assistant_message": {
                "role": "assistant",
                "content": "",
                "tool_calls": [call],
            },
        }

    monkeypatch.setattr(loop_mod, "chat", chat)
    result = Agent(
        use_skills=False, allow_delegate=False, max_turns=5, workspace=tmp_path
    ).run("Write the classification cell")

    assert any("Permission denied" in text for text in observations)
    assert any("not backed by this run's ledger" in text for text in observations)
    assert result["stop_reason"] == "submitted"
    assert result["submitted_output"]["completion_bullets"] == [
        "Explained the permission limitation"
    ]