"""Real ACP editor permissions over an ordinary daemon's canonical ACP route."""
import asyncio
from contextlib import suppress
from http.server import BaseHTTPRequestHandler
import json

import pytest

from tests.acp_adapter.test_gateway_sessions import editor, viewer
from tests.waiting import async_wait_until


class EditPeer(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        request = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
        self.server.requests.append(request)
        # write_file refuses to replace an existing file this task never read in full
        # (stale-overwrite guard), so the scripted model reads first, then edits.
        # Counted since the latest user turn, so a second prompt in the session edits again.
        messages = request.get('messages', [])
        last_user = max((i for i, m in enumerate(messages) if m['role'] == 'user'), default=-1)
        seen = sum(1 for m in messages[last_user + 1:] if m['role'] == 'tool')
        tool = seen < 2
        message = {'role': 'assistant', 'content': 'EDITOR_FINISHED'}
        if seen == 0:
            message = {'role': 'assistant', 'content': None, 'tool_calls': [{
                'id': 'owned-read', 'type': 'function', 'function': {
                    'name': 'read_file', 'arguments': json.dumps({'path': str(self.server.target)})}}]}
        elif seen == 1:
            message = {'role': 'assistant', 'content': None, 'tool_calls': [{
                'id': 'owned-edit', 'type': 'function', 'function': {
                    'name': 'write_file', 'arguments': json.dumps({
                        'path': str(self.server.target), 'content': 'approved edit'})}}]}
        self.send_response(200)
        self.send_header('Content-Type', 'text/event-stream' if request.get('stream') else 'application/json')
        self.end_headers()
        if request.get('stream'):
            if tool:
                message['tool_calls'][0]['index'] = 0
            for delta, finish in [(message, None), ({}, 'tool_calls' if tool else 'stop')]:
                self.wfile.write(('data: ' + json.dumps({'id': 'edit', 'object': 'chat.completion.chunk',
                    'created': 1, 'model': 'fixture', 'choices': [{'index': 0, 'delta': delta,
                    'finish_reason': finish}]}) + '\n\n').encode())
            self.wfile.write(b'data: [DONE]\n\n')
        else:
            self.wfile.write(json.dumps({'id': 'edit', 'object': 'chat.completion', 'created': 1,
                'model': 'fixture', 'choices': [{'index': 0, 'message': message,
                'finish_reason': 'tool_calls' if tool else 'stop'}]}).encode())
        self.wfile.flush()


@pytest.mark.platforms("linux")
@pytest.mark.asyncio
async def test_acp_load_from_another_folder_moves_the_session_there(daemon, tmp_path):
    """Main rebound a session opened from another folder (#134889 ACP-12): the owner's workspace
    mutation moves the authoritative cwd, which every viewer then reads."""
    other = tmp_path / 'other'
    other.mkdir()
    async with viewer(daemon) as ws:
        created = await ws.rpc('session.create', source='acp', cwd=str(tmp_path), request_id='cwd')
        sid = created['session_id']
        assert created['info']['cwd'] == str(tmp_path)
        async with editor(daemon, tmp_path) as acp:
            await acp.rpc('initialize', protocolVersion=1, clientCapabilities={})
            loaded = await acp.rpc('session/load', sessionId=sid, cwd=str(other), mcpServers=[])
            assert 'result' in loaded, loaded
            assert (await ws.rpc('session.info', session_id=sid))['cwd'] == str(other)
            resumed = await acp.rpc('session/resume', sessionId=sid, cwd=str(tmp_path), mcpServers=[])
            assert 'result' in resumed, resumed
            assert (await ws.rpc('session.info', session_id=sid))['cwd'] == str(tmp_path)
            forked = await acp.rpc('session/fork', sessionId=sid, cwd=str(other), mcpServers=[])
            assert 'result' in forked, forked
            child = forked['result']['sessionId']
            assert (await ws.rpc('session.info', session_id=child))['cwd'] == str(other)
            assert (await ws.rpc('session.info', session_id=sid))['cwd'] == str(tmp_path)


@pytest.mark.platforms("linux")
@pytest.mark.asyncio
async def test_real_editor_diff_requires_native_consent_before_write(daemon, tmp_path, model_peer):
    target = tmp_path / 'owned.txt'
    target.write_text('original edit')
    model_peer.target = target
    model_peer.RequestHandlerClass = EditPeer
    async with viewer(daemon) as ws:
        created = await ws.rpc('session.create', source='acp', cwd=str(tmp_path), request_id='editor-diff')
        sid = created['session_id']
        async with editor(daemon, tmp_path) as acp:
            await acp.rpc('initialize', protocolVersion=1, clientCapabilities={})
            loaded = await acp.rpc('session/load', sessionId=sid, cwd=str(tmp_path), mcpServers=[])
            assert 'result' in loaded, loaded
            task = asyncio.create_task(acp.rpc('session/prompt', sessionId=sid,
                prompt=[{'type': 'text', 'text': 'Edit the owned fixture'}]))
            try:
                await async_wait_until(lambda: any(f.get('method') == 'session/request_permission' for f in acp.frames),
                                       timeout=20, what='the edit permission request')
                permission = next(f for f in acp.frames if f.get('method') == 'session/request_permission')
                call = permission['params']['toolCall']
                assert call['kind'] == 'edit', permission
                assert call['content'][0]['oldText'] == 'original edit'
                assert call['content'][0]['newText'] == 'approved edit'
                assert target.read_text() == 'original edit'
                acp.process.stdin.write(json.dumps({'jsonrpc': '2.0', 'id': permission['id'], 'result': {
                    'outcome': {'outcome': 'selected', 'optionId': 'allow_once'}}}).encode() + b'\n')
                await acp.process.stdin.drain()
                assert (await task)['result']['stopReason'] == 'end_turn'
                assert target.read_text() == 'approved edit'
                print('ACP_EDITOR_RECEIPT=' + json.dumps({'session_id': sid, 'permission': permission,
                    'owned_write_after_consent': True}))
            finally:
                task.cancel()
                with suppress(asyncio.CancelledError):
                    await task


@pytest.mark.platforms("linux")
@pytest.mark.asyncio
async def test_accept_edits_runs_a_workspace_edit_unasked_and_a_sensitive_one_still_asks(daemon, tmp_path, model_peer):
    """ACP-7: Accept Edits is the session's ``edit_policy`` on the owner; a second surface reads it, a
    workspace edit runs without a permission request, and a sensitive path (``.env``) still asks."""
    target = tmp_path / 'owned.txt'
    target.write_text('original edit')
    model_peer.target = target
    model_peer.RequestHandlerClass = EditPeer
    async with viewer(daemon) as ws:
        sid = (await ws.rpc('session.create', source='acp', cwd=str(tmp_path), request_id='accept'))['session_id']
        assert (await ws.rpc('session.info', session_id=sid))['edit_approval_policy'] == 'ask'
        async with editor(daemon, tmp_path) as acp:
            await acp.rpc('initialize', protocolVersion=1, clientCapabilities={})
            loaded = await acp.rpc('session/load', sessionId=sid, cwd=str(tmp_path), mcpServers=[])
            modes = loaded['result']['modes']
            assert modes['currentModeId'] == 'default', loaded
            assert [m['id'] for m in modes['availableModes']] == ['default', 'accept_edits', 'dont_ask']
            assert 'result' in await acp.rpc('session/set_mode', sessionId=sid, modeId='accept_edits')
            # Every surface attached to the session sees the same policy.
            assert (await ws.rpc('session.info', session_id=sid))['edit_approval_policy'] == 'workspace_session'
            turn = await acp.rpc('session/prompt', sessionId=sid, prompt=[{'type': 'text', 'text': 'Edit it'}])
            assert turn['result']['stopReason'] == 'end_turn', turn
            assert target.read_text() == 'approved edit'
            assert not any(f.get('method') == 'session/request_permission' for f in acp.frames)

            sensitive = tmp_path / '.env'
            sensitive.write_text('SECRET=1')
            model_peer.target = sensitive
            task = asyncio.create_task(acp.rpc('session/prompt', sessionId=sid,
                prompt=[{'type': 'text', 'text': 'Edit the env file'}]))
            try:
                await async_wait_until(lambda: any(f.get('method') == 'session/request_permission' for f in acp.frames),
                                       timeout=20, what='the sensitive edit permission request')
                permission = next(f for f in acp.frames if f.get('method') == 'session/request_permission')
                assert permission['params']['toolCall']['kind'] == 'edit', permission
                acp.process.stdin.write(json.dumps({'jsonrpc': '2.0', 'id': permission['id'], 'result': {
                    'outcome': {'outcome': 'selected', 'optionId': 'deny'}}}).encode() + b'\n')
                await acp.process.stdin.drain()
                assert (await task)['result']['stopReason'] == 'end_turn'
                assert sensitive.read_text() == 'SECRET=1'
            finally:
                task.cancel()
                with suppress(asyncio.CancelledError):
                    await task
            # A second editor attaching to the session is shown the session's mode.
            async with editor(daemon, tmp_path) as second:
                await second.rpc('initialize', protocolVersion=1, clientCapabilities={})
                again = await second.rpc('session/load', sessionId=sid, cwd=str(tmp_path), mcpServers=[])
                assert again['result']['modes']['currentModeId'] == 'accept_edits', again