zk-data-agent/tests/test_agent_runtime.py

from __future__ import annotations

import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from src.agent_tool_core import ToolStreamUpdate
from src.agent_session import AgentMessage
from src.agent_runtime import LocalCodingAgent, MAX_AUTO_CONTINUATIONS
from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
from src.agent_types import (
    AgentPermissions,
    AgentRuntimeConfig,
    BudgetConfig,
    ModelConfig,
    OutputSchemaConfig,
    ToolExecutionResult,
    UsageStats,
)
from src.compact import CompactionResult
from src.openai_compat import OpenAICompatClient
from src.session_store import (
    StoredAgentSession,
    load_agent_session,
    serialize_model_config,
    serialize_runtime_config,
)
from src.token_budget import TokenBudgetSnapshot


class FakeHTTPResponse:
    def __init__(self, payload: dict[str, object]) -> None:
        self.payload = payload

    def read(self) -> bytes:
        return json.dumps(self.payload).encode('utf-8')

    def __enter__(self) -> 'FakeHTTPResponse':
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        return None


class FakeStreamingHTTPResponse:
    def __init__(self, payloads: list[dict[str, object]]) -> None:
        self.lines: list[bytes] = []
        for payload in payloads:
            chunk = f'data: {json.dumps(payload)}\n\n'
            self.lines.extend(part.encode('utf-8') for part in chunk.splitlines(keepends=True))
        done_chunk = 'data: [DONE]\n\n'
        self.lines.extend(part.encode('utf-8') for part in done_chunk.splitlines(keepends=True))

    def readline(self) -> bytes:
        if not self.lines:
            return b''
        return self.lines.pop(0)

    def __enter__(self) -> 'FakeStreamingHTTPResponse':
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        return None


def make_urlopen_side_effect(responses: list[dict[str, object]]):
    queued = [FakeHTTPResponse(payload) for payload in responses]

    def _fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
        return queued.pop(0)

    return _fake_urlopen


def make_recording_urlopen_side_effect(
    responses: list[dict[str, object]],
    recorded_payloads: list[dict[str, object]],
):
    queued = [FakeHTTPResponse(payload) for payload in responses]

    def _fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
        body = request_obj.data.decode('utf-8')
        recorded_payloads.append(json.loads(body))
        return queued.pop(0)

    return _fake_urlopen


def make_streaming_urlopen_side_effect(
    responses: list[list[dict[str, object]]],
):
    queued = [FakeStreamingHTTPResponse(payloads) for payloads in responses]

    def _fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
        return queued.pop(0)

    return _fake_urlopen


def make_recording_streaming_urlopen_side_effect(
    responses: list[list[dict[str, object]]],
    recorded_payloads: list[dict[str, object]],
):
    queued = [FakeStreamingHTTPResponse(payloads) for payloads in responses]

    def _fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
        body = request_obj.data.decode('utf-8')
        recorded_payloads.append(json.loads(body))
        return queued.pop(0)

    return _fake_urlopen


class AgentRuntimeTests(unittest.TestCase):
    def test_custom_project_agent_is_resolved_for_child_agent_configuration(self) -> None:
        with tempfile.TemporaryDirectory() as home_dir, tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            agents_dir = workspace / '.claude' / 'agents'
            agents_dir.mkdir(parents=True)
            (agents_dir / 'Explore.md').write_text(
                (
                    '---\n'
                    'name: Explore\n'
                    'description: "Project-specific explore agent."\n'
                    'tools: read_file, grep_search\n'
                    'model: child-model\n'
                    'initialPrompt: Start with a repository scan.\n'
                    '---\n\n'
                    'Search the repository and report findings.\n'
                ),
                encoding='utf-8',
            )
            with patch.dict(os.environ, {'HOME': home_dir}):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(model='parent-model'),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                agent_def = agent._resolve_agent_definition({'subagent_type': 'Explore'})
                child_model = agent._resolve_child_model_config({}, agent_def)
                child_tools = agent._filter_tools_for_agent(agent_def)
            self.assertEqual(agent_def.source, 'projectSettings')
            self.assertEqual(agent_def.initial_prompt, 'Start with a repository scan.')
            self.assertEqual(child_model.model, 'child-model')
            self.assertEqual(sorted(child_tools), ['grep_search', 'read_file'])

    def test_builtin_child_model_alias_inherits_non_claude_parent_model(self) -> None:
        agent = LocalCodingAgent(
            model_config=ModelConfig(model='xiaomi/mimo-v2.5-pro'),
            runtime_config=AgentRuntimeConfig(cwd=Path.cwd()),
        )
        agent_def = agent._resolve_agent_definition({'subagent_type': 'Explore'})
        child_model = agent._resolve_child_model_config({}, agent_def)

        self.assertEqual(agent_def.model, 'haiku')
        self.assertEqual(child_model.model, 'xiaomi/mimo-v2.5-pro')

    def test_openai_client_parses_tool_calls(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Inspecting the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "hello.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            }
        ]
        with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
            client = OpenAICompatClient(
                ModelConfig(
                    model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                    base_url='http://127.0.0.1:8000/v1',
                )
            )
            turn = client.complete(
                messages=[{'role': 'user', 'content': 'read hello.txt'}],
                tools=[],
            )
        self.assertEqual(turn.content, 'Inspecting the file.')
        self.assertEqual(len(turn.tool_calls), 1)
        self.assertEqual(turn.tool_calls[0].name, 'read_file')
        self.assertEqual(turn.tool_calls[0].arguments['path'], 'hello.txt')

    def test_openai_client_streams_content_and_usage(self) -> None:
        responses = [
            [
                {'choices': [{'delta': {'content': 'Hello '}, 'finish_reason': None}]},
                {'choices': [{'delta': {'content': 'world'}, 'finish_reason': None}]},
                {
                    'choices': [{'delta': {}, 'finish_reason': 'stop'}],
                    'usage': {'prompt_tokens': 10, 'completion_tokens': 3},
                },
            ]
        ]
        with patch(
            'src.openai_compat.request.urlopen',
            side_effect=make_streaming_urlopen_side_effect(responses),
        ):
            client = OpenAICompatClient(
                ModelConfig(
                    model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                    base_url='http://127.0.0.1:8000/v1',
                )
            )
            events = list(
                client.stream(
                    messages=[{'role': 'user', 'content': 'say hello'}],
                    tools=[],
                )
            )
        self.assertEqual(events[0].type, 'message_start')
        self.assertEqual(
            ''.join(event.delta for event in events if event.type == 'content_delta'),
            'Hello world',
        )
        usage_events = [event for event in events if event.type == 'usage']
        self.assertEqual(len(usage_events), 1)
        self.assertEqual(usage_events[0].usage.input_tokens, 10)
        self.assertEqual(usage_events[0].usage.output_tokens, 3)

    def test_agent_executes_tool_calls_against_fake_backend(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will inspect the file first.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "hello.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'The file contains hello world.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8')
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Inspect hello.txt')

        self.assertEqual(result.final_output, 'The file contains hello world.')
        self.assertEqual(result.tool_calls, 1)
        transcript_roles = [entry.get('role') for entry in result.transcript]
        self.assertEqual(transcript_roles, ['user', 'assistant', 'tool', 'assistant'])
        self.assertIn('I will inspect the file first.', result.transcript[1]['content'])
        self.assertIn('hello world', result.transcript[2]['content'])
        self.assertIn('The file contains hello world.', result.transcript[3]['content'])
        self.assertGreaterEqual(len(result.file_history), 0)

    def test_runtime_guidance_before_tools_skips_stale_tool_plan(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will inspect the old file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "old.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Replanned with the guidance.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        provider_calls = 0

        def guidance_provider() -> tuple[dict[str, object], ...]:
            nonlocal provider_calls
            provider_calls += 1
            if provider_calls == 2:
                return ({'id': 7, 'content': 'Use new.txt instead.'},)
            return ()

        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'old.txt').write_text('old content\n', encoding='utf-8')
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                agent.runtime_guidance_provider = guidance_provider
                result = agent.run('Inspect old.txt')

        self.assertEqual(result.final_output, 'Replanned with the guidance.')
        self.assertEqual(result.tool_calls, 0)
        self.assertFalse(result.file_history)
        replan_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_replan_before_tools'
        ]
        self.assertEqual(len(replan_events), 1)
        injected_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_injected'
        ]
        self.assertEqual(injected_events[-1].get('stage'), 'before_tools')
        self.assertEqual(len(recorded_payloads), 2)
        second_messages = recorded_payloads[1]['messages']
        assert isinstance(second_messages, list)
        roles = [
            message.get('role')
            for message in second_messages
            if isinstance(message, dict)
        ]
        self.assertEqual(roles[-3:], ['assistant', 'tool', 'user'])
        self.assertIn('runtime_guidance_replan', str(second_messages[-2]))
        self.assertIn('Use new.txt instead.', str(second_messages[-1]))

    def test_runtime_guidance_interrupts_running_streaming_tool(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will run the long tool.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'long_tool',
                                        'arguments': '{}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Replanned after the interruption.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        provider_calls = 0
        cancel_observed_after_guidance: list[bool] = []

        def guidance_provider() -> tuple[dict[str, object], ...]:
            nonlocal provider_calls
            provider_calls += 1
            if provider_calls == 3:
                return ({'id': 12, 'content': '停一下，改成只生成 3 条。'},)
            return ()

        def fake_execute_tool_streaming(registry, name, arguments, context):  # noqa: ANN001
            self.assertEqual(name, 'long_tool')
            yield ToolStreamUpdate(kind='delta', content='working\n', stream='stdout')
            cancel_observed_after_guidance.append(bool(context.cancel_event.is_set()))
            yield ToolStreamUpdate(
                kind='result',
                result=ToolExecutionResult(
                    name=name,
                    ok=False,
                    content='Tool saw cancellation.',
                    metadata={'cancel_seen': bool(context.cancel_event.is_set())},
                ),
            )

        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with (
                patch(
                    'src.openai_compat.request.urlopen',
                    side_effect=make_recording_urlopen_side_effect(
                        responses,
                        recorded_payloads,
                    ),
                ),
                patch(
                    'src.agent_runtime.execute_tool_streaming',
                    side_effect=fake_execute_tool_streaming,
                ),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                agent.runtime_guidance_provider = guidance_provider
                result = agent.run('Run a long tool')

        self.assertEqual(result.final_output, 'Replanned after the interruption.')
        self.assertEqual(result.tool_calls, 1)
        self.assertEqual(cancel_observed_after_guidance, [True])
        interrupt_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_interrupt_tool'
        ]
        self.assertEqual(len(interrupt_events), 1)
        injected_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_injected'
        ]
        self.assertEqual(injected_events[-1].get('stage'), 'during_tool_interrupted')
        self.assertEqual(len(recorded_payloads), 2)
        second_messages = recorded_payloads[1]['messages']
        assert isinstance(second_messages, list)
        roles = [
            message.get('role')
            for message in second_messages
            if isinstance(message, dict)
        ]
        self.assertEqual(roles[-3:], ['assistant', 'tool', 'user'])
        self.assertIn('runtime_guidance_interrupted_tool', str(second_messages[-2]))
        self.assertIn('改成只生成 3 条', str(second_messages[-1]))

    def test_runtime_guidance_during_tool_can_defer_until_tool_result(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will run the long tool.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'long_tool',
                                        'arguments': '{}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Replanned after the tool result.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        provider_calls = 0
        cancel_observed_after_guidance: list[bool] = []

        def guidance_provider() -> tuple[dict[str, object], ...]:
            nonlocal provider_calls
            provider_calls += 1
            if provider_calls == 3:
                return ({'id': 13, 'content': '完成后顺便整理成 Markdown。'},)
            return ()

        def fake_execute_tool_streaming(registry, name, arguments, context):  # noqa: ANN001
            self.assertEqual(name, 'long_tool')
            yield ToolStreamUpdate(kind='delta', content='working\n', stream='stdout')
            cancel_observed_after_guidance.append(bool(context.cancel_event.is_set()))
            yield ToolStreamUpdate(
                kind='result',
                result=ToolExecutionResult(
                    name=name,
                    ok=True,
                    content='Tool completed.',
                    metadata={'cancel_seen': bool(context.cancel_event.is_set())},
                ),
            )

        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with (
                patch(
                    'src.openai_compat.request.urlopen',
                    side_effect=make_recording_urlopen_side_effect(
                        responses,
                        recorded_payloads,
                    ),
                ),
                patch(
                    'src.agent_runtime.execute_tool_streaming',
                    side_effect=fake_execute_tool_streaming,
                ),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                agent.runtime_guidance_provider = guidance_provider
                result = agent.run('Run a long tool')

        self.assertEqual(result.final_output, 'Replanned after the tool result.')
        self.assertEqual(cancel_observed_after_guidance, [False])
        deferred_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_deferred_after_tool'
        ]
        self.assertEqual(len(deferred_events), 1)
        interrupt_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_interrupt_tool'
        ]
        self.assertFalse(interrupt_events)
        injected_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_injected'
        ]
        self.assertEqual(injected_events[-1].get('stage'), 'after_tool')
        second_messages = recorded_payloads[1]['messages']
        assert isinstance(second_messages, list)
        self.assertIn('runtime_guidance_deferred_after_tool', str(second_messages[-2]))
        self.assertIn('完成后顺便整理成 Markdown', str(second_messages[-1]))

    def test_agent_persists_max_turns_explanation_after_tool_result(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': '我先读取文件。',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "hello.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8')
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace, max_turns=1),
                )
                result = agent.run('Inspect hello.txt')

        self.assertEqual(result.stop_reason, 'max_turns')
        self.assertIn('最大步骤上限', result.final_output)
        self.assertEqual(result.transcript[-1]['role'], 'assistant')
        self.assertEqual(result.transcript[-1]['stop_reason'], 'max_turns')
        self.assertIn('最大步骤上限', result.transcript[-1]['content'])

    def test_runtime_guidance_before_finish_continues_instead_of_finishing(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Draft answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Revised final answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        provider_calls = 0

        def guidance_provider() -> tuple[dict[str, object], ...]:
            nonlocal provider_calls
            provider_calls += 1
            if provider_calls == 2:
                return ({'id': 9, 'content': 'Add the timing summary before finalizing.'},)
            return ()

        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                agent.runtime_guidance_provider = guidance_provider
                result = agent.run('Prepare an answer')

        self.assertEqual(result.final_output, 'Revised final answer.')
        injected_events = [
            event
            for event in result.events
            if event.get('type') == 'runtime_guidance_injected'
        ]
        self.assertEqual(injected_events[-1].get('stage'), 'before_finish')
        self.assertEqual(len(recorded_payloads), 2)
        second_messages = recorded_payloads[1]['messages']
        assert isinstance(second_messages, list)
        self.assertEqual(second_messages[-2].get('role'), 'assistant')
        self.assertEqual(second_messages[-2].get('content'), 'Draft answer.')
        self.assertEqual(second_messages[-1].get('role'), 'user')
        self.assertIn('Add the timing summary', str(second_messages[-1].get('content')))

    def test_agent_stops_after_data_agent_goal_requires_review(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will prepare a reviewable generation goal.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'data_agent_prepare_generation_goal',
                                        'arguments': json.dumps(
                                            {
                                                'dataset_label': '地图和生活边界数据',
                                                'goal_summary': '生成附近生活服务边界数据',
                                                'target': 'Agent(tag="life_service")',
                                                'plan_hint': '建议先生成 10 条单轮，输出到 tasks/demo/records.jsonl；具体数量、轮次和路径在 generation plan 中确认。',
                                                'coverage': '附近吃喝玩乐',
                                                'exclusions': '不要生成导航路线类 query',
                                                'source_refs': ['manual:user'],
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Prepare data generation goal')

        self.assertEqual(result.stop_reason, 'user_review_required')
        self.assertEqual(result.tool_calls, 1)
        self.assertIn('先确认边界', result.final_output)
        self.assertIn('goal_id', result.final_output)
        self.assertIn('下一步', result.final_output)
        self.assertIn('建议先生成 10 条单轮', result.final_output)
        review_messages = [
            entry
            for entry in result.transcript
            if entry.get('role') == 'assistant'
            and entry.get('stop_reason') == 'user_review_required'
        ]
        self.assertEqual(len(review_messages), 1)
        self.assertIn('生成目标', str(review_messages[0].get('content', '')))
        self.assertTrue(
            any(event.get('type') == 'user_review_required' for event in result.events)
        )
        self.assertTrue(
            any(
                event.get('type') == 'user_review_required'
                and isinstance(event.get('review_message_id'), str)
                for event in result.events
            )
        )

    def test_agent_appends_review_message_after_data_agent_plan_requires_review(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            context = build_tool_context(AgentRuntimeConfig(cwd=workspace))
            goal_result = execute_tool(
                default_tool_registry(),
                'data_agent_prepare_generation_goal',
                {
                    'dataset_label': '地图和生活边界数据',
                    'goal_summary': '生成附近生活服务边界数据',
                    'target': 'Agent(tag="life_service")',
                    'coverage': '附近吃喝玩乐',
                    'exclusions': '不要生成导航路线类 query',
                    'source_refs': ['manual:user'],
                },
                context,
            )
            self.assertTrue(goal_result.ok, goal_result.content)
            goal_id = json.loads(goal_result.content)['goal']['goal_id']
            confirm_result = execute_tool(
                default_tool_registry(),
                'data_agent_confirm_generation_goal',
                {'goal_id': goal_id, 'confirmation': '确认目标', 'reviewed_revision': 1},
                context,
            )
            self.assertTrue(confirm_result.ok, confirm_result.content)
            responses = [
                {
                    'choices': [
                        {
                            'message': {
                                'role': 'assistant',
                                'content': 'I will prepare a reviewable generation plan.',
                                'tool_calls': [
                                    {
                                        'id': 'call_1',
                                        'type': 'function',
                                        'function': {
                                            'name': 'data_agent_prepare_generation_plan',
                                            'arguments': json.dumps(
                                                {
                                                    'confirmed_goal_id': goal_id,
                                                    'dataset_label': '地图和生活边界数据',
                                                    'target': 'Agent(tag="life_service")',
                                                    'total_count': 10,
                                                    'turn_mix': '10 条单轮',
                                                    'coverage': '附近吃喝玩乐',
                                                    'exclusions': '不要生成导航路线类 query',
                                                    'output_path': 'tasks/demo/records.jsonl',
                                                }
                                            ),
                                        },
                                    }
                                ],
                            },
                            'finish_reason': 'tool_calls',
                        }
                    ]
                }
            ]
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Prepare data generation plan')

        self.assertEqual(result.stop_reason, 'user_review_required')
        self.assertIn('只补充生成参数', result.final_output)
        self.assertIn('plan_id', result.final_output)
        self.assertNotIn('target_definitions', result.final_output)
        review_messages = [
            entry
            for entry in result.transcript
            if entry.get('role') == 'assistant'
            and entry.get('stop_reason') == 'user_review_required'
        ]
        self.assertEqual(len(review_messages), 1)
        self.assertIn('生成参数', str(review_messages[0].get('content', '')))

    def test_agent_stops_after_ask_user_question_requires_review(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I need the user to choose a strategy.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'ask_user_question',
                                        'arguments': json.dumps(
                                            {
                                                'question': '选择线上挖掘策略？',
                                                'choices': ['先宽召回', '先精确过滤'],
                                                'allow_free_text': True,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ]
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Ask before mining')

        self.assertEqual(result.stop_reason, 'user_review_required')
        self.assertEqual(result.tool_calls, 1)
        self.assertIn('选择线上挖掘策略', result.final_output)
        self.assertIn('先宽召回', result.final_output)
        review_messages = [
            entry
            for entry in result.transcript
            if entry.get('role') == 'assistant'
            and entry.get('stop_reason') == 'user_review_required'
        ]
        self.assertEqual(len(review_messages), 1)
        self.assertTrue(
            any(
                event.get('type') == 'user_review_required'
                and event.get('tool_name') == 'ask_user_question'
                for event in result.events
            )
        )

    def test_write_tool_is_blocked_without_permission(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            config = AgentRuntimeConfig(cwd=Path(tmp_dir))
            context = build_tool_context(config)
            result = execute_tool(
                default_tool_registry(),
                'write_file',
                {'path': 'blocked.txt', 'content': 'data'},
                context,
            )
        self.assertFalse(result.ok)
        self.assertIn('--allow-write', result.content)
        self.assertEqual(result.metadata.get('error_kind'), 'permission_denied')

    def test_build_tool_context_supports_safe_env_overlay_for_shell_tools(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            config = AgentRuntimeConfig(
                cwd=Path(tmp_dir),
                permissions=AgentPermissions(allow_shell_commands=True),
            )
            context = build_tool_context(
                config,
                extra_env={'HOOK_SAFE_TOKEN': 'demo-secret'},
            )
            result = execute_tool(
                default_tool_registry(),
                'bash',
                {'command': 'printf %s "$HOOK_SAFE_TOKEN"'},
                context,
            )
        self.assertTrue(result.ok)
        self.assertIn('demo-secret', result.content)

    def test_local_slash_command_returns_without_model_call(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            agent = LocalCodingAgent(
                model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
                runtime_config=AgentRuntimeConfig(cwd=Path(tmp_dir)),
            )
            result = agent.run('/permissions')
        self.assertEqual(result.turns, 0)
        self.assertEqual(result.tool_calls, 0)
        self.assertIn('# Permissions', result.final_output)

    def test_slash_skills_lists_bundled_skills(self) -> None:
        from src.bundled_skills import get_bundled_skills

        with tempfile.TemporaryDirectory() as tmp_dir:
            agent = LocalCodingAgent(
                model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
                runtime_config=AgentRuntimeConfig(cwd=Path(tmp_dir)),
            )
            result = agent.run('/skills')
        self.assertEqual(result.turns, 0)
        self.assertEqual(result.tool_calls, 0)
        self.assertIn('Available Skills', result.final_output)
        for skill in get_bundled_skills():
            if skill.user_invocable:
                self.assertIn(skill.name, result.final_output)
        # Slash command names (e.g. 'permissions') must NOT appear — that was
        # the old buggy behavior the upstream `/skills` command never did.
        self.assertNotIn('/permissions', result.final_output)

    def test_clear_runtime_state_drops_session_env_vars(self) -> None:
        from src.session_env_vars import (
            clear_session_env_vars,
            get_session_env_vars,
            set_session_env_var,
        )

        clear_session_env_vars()
        try:
            with tempfile.TemporaryDirectory() as tmp_dir:
                agent = LocalCodingAgent(
                    model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
                    runtime_config=AgentRuntimeConfig(cwd=Path(tmp_dir)),
                )
                set_session_env_var('CLAW_CLEAR_TEST', 'before')
                self.assertEqual(get_session_env_vars()['CLAW_CLEAR_TEST'], 'before')
                agent.clear_runtime_state()
                self.assertNotIn('CLAW_CLEAR_TEST', get_session_env_vars())
        finally:
            clear_session_env_vars()

    def test_agent_persists_session_and_can_resume(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Initial answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Continued answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ]
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            runtime_config = AgentRuntimeConfig(
                cwd=workspace,
                session_directory=session_dir,
            )
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=runtime_config,
                )
                first_result = agent.run('Start task')
                self.assertIsNotNone(first_result.session_id)
                stored = load_agent_session(first_result.session_id or '', directory=session_dir)

                resumed_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=runtime_config,
                )
                second_result = resumed_agent.resume('Continue the task', stored)

                self.assertTrue((session_dir / f'{first_result.session_id}.json').exists())

        self.assertEqual(first_result.final_output, 'Initial answer.')
        self.assertEqual(second_result.final_output, 'Continued answer.')
        self.assertEqual(second_result.session_id, first_result.session_id)
        self.assertEqual(len(recorded_payloads), 2)
        resumed_messages = recorded_payloads[1]['messages']
        assert isinstance(resumed_messages, list)
        contents = [message.get('content') for message in resumed_messages if isinstance(message, dict)]
        self.assertIn('Start task', contents)
        self.assertIn('Initial answer.', contents)
        self.assertIn('Continue the task', contents)

    def test_agent_streams_runtime_output_and_usage(self) -> None:
        responses = [
            [
                {'choices': [{'delta': {'content': 'Streaming '}, 'finish_reason': None}]},
                {'choices': [{'delta': {'content': 'works.'}, 'finish_reason': None}]},
                {
                    'choices': [{'delta': {}, 'finish_reason': 'stop'}],
                    'usage': {'prompt_tokens': 14, 'completion_tokens': 5},
                },
            ]
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_streaming_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        stream_model_responses=True,
                    ),
                )
                result = agent.run('Say streaming works')
        self.assertEqual(result.final_output, 'Streaming works.')
        self.assertEqual(result.usage.input_tokens, 14)
        self.assertEqual(result.usage.output_tokens, 5)
        self.assertTrue(any(event.get('type') == 'content_delta' for event in result.events))
        self.assertIsNotNone(result.scratchpad_directory)
        assert result.scratchpad_directory is not None
        self.assertTrue(Path(result.scratchpad_directory).is_dir())
        assistant_messages = [
            message for message in result.transcript if message.get('role') == 'assistant'
        ]
        self.assertEqual(len(assistant_messages), 1)
        metadata = assistant_messages[0].get('metadata', {})
        mutation_totals = metadata.get('mutation_totals', {})
        self.assertGreaterEqual(mutation_totals.get('assistant_delta_append', 0), 2)
        self.assertEqual(mutation_totals.get('assistant_finalize', 0), 1)

    def test_agent_streams_tool_calls_and_reconstructs_arguments(self) -> None:
        responses = [
            [
                {
                    'choices': [
                        {
                            'delta': {
                                'tool_calls': [
                                    {
                                        'index': 0,
                                        'id': 'call_1',
                                        'function': {
                                            'name': 'read_file',
                                            'arguments': '{"path": "hello',
                                        },
                                    }
                                ]
                            },
                            'finish_reason': None,
                        }
                    ]
                },
                {
                    'choices': [
                        {
                            'delta': {
                                'tool_calls': [
                                    {
                                        'index': 0,
                                        'function': {
                                            'arguments': '.txt"}',
                                        },
                                    }
                                ]
                            },
                            'finish_reason': None,
                        }
                    ]
                },
                {
                    'choices': [{'delta': {}, 'finish_reason': 'tool_calls'}],
                    'usage': {'prompt_tokens': 9, 'completion_tokens': 4},
                },
            ],
            [
                {'choices': [{'delta': {'content': 'Read done.'}, 'finish_reason': None}]},
                {
                    'choices': [{'delta': {}, 'finish_reason': 'stop'}],
                    'usage': {'prompt_tokens': 11, 'completion_tokens': 2},
                },
            ],
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8')
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_streaming_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        stream_model_responses=True,
                    ),
                )
                result = agent.run('Inspect hello.txt')
        self.assertEqual(result.final_output, 'Read done.')
        self.assertEqual(result.tool_calls, 1)
        self.assertEqual(result.usage.input_tokens, 20)
        self.assertEqual(result.usage.output_tokens, 6)
        assistant_messages = [message for message in result.transcript if message.get('role') == 'assistant']
        self.assertTrue(any(message.get('tool_calls') for message in assistant_messages))
        assistant_with_tool = next(
            message
            for message in assistant_messages
            if message.get('tool_calls')
        )
        metadata = assistant_with_tool.get('metadata', {})
        mutation_totals = metadata.get('mutation_totals', {})
        self.assertGreaterEqual(mutation_totals.get('assistant_tool_call_delta', 0), 2)
        self.assertEqual(mutation_totals.get('assistant_finalize', 0), 1)

    def test_agent_reports_malformed_streaming_tool_arguments(self) -> None:
        responses = [
            [
                {
                    'choices': [
                        {
                            'delta': {
                                'tool_calls': [
                                    {
                                        'index': 0,
                                        'id': 'call_1',
                                        'function': {
                                            'name': 'data_agent_normalize_dataset_draft',
                                            'arguments': '{"draft_text": "target: Agent(tag="地图导航")"}',
                                        },
                                    }
                                ]
                            },
                            'finish_reason': None,
                        }
                    ]
                },
                {
                    'choices': [{'delta': {}, 'finish_reason': 'tool_calls'}],
                    'usage': {'prompt_tokens': 9, 'completion_tokens': 4},
                },
            ]
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_streaming_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        stream_model_responses=True,
                    ),
                )
                result = agent.run('Generate records')

        self.assertEqual(result.stop_reason, 'backend_error')
        self.assertIn('模型返回的工具参数不是合法 JSON', result.final_output)
        self.assertIn('data_agent_normalize_dataset_draft', result.final_output)
        malformed = [
            message
            for message in result.transcript
            if message.get('stop_reason') == 'malformed_tool_arguments'
        ]
        self.assertEqual(len(malformed), 1)
        self.assertNotIn('tool_calls', malformed[0])
        self.assertEqual(result.transcript[-1]['content'], result.final_output)

    def test_transcript_entries_include_structured_blocks(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will inspect the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "hello.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Done reading.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8')
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Inspect hello.txt')
        assistant_with_tool = next(
            message
            for message in result.transcript
            if message.get('role') == 'assistant' and message.get('tool_calls')
        )
        self.assertIn('blocks', assistant_with_tool)
        block_types = [block.get('type') for block in assistant_with_tool['blocks']]
        self.assertIn('text', block_types)
        self.assertIn('tool_call', block_types)
        tool_message = next(message for message in result.transcript if message.get('role') == 'tool')
        self.assertIn('blocks', tool_message)
        self.assertEqual(tool_message['blocks'][0]['type'], 'tool_result')

    def test_agent_inserts_compact_boundary_when_threshold_is_exceeded(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will inspect the file and then continue.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "hello.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 30, 'completion_tokens': 10},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Compaction test completed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 12, 'completion_tokens': 4},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8')
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        auto_compact_threshold_tokens=80,
                        compact_preserve_messages=1,
                    ),
                )
                result = agent.run(
                    'Read hello.txt and then continue with a detailed explanation that is intentionally long enough to trigger compaction.'
                )
        self.assertEqual(result.final_output, 'Compaction test completed.')
        compact_events = [event for event in result.events if event.get('type') == 'compact_boundary']
        self.assertEqual(len(compact_events), 1)
        second_request_messages = recorded_payloads[1]['messages']
        assert isinstance(second_request_messages, list)
        compact_messages = [
            message for message in second_request_messages
            if isinstance(message, dict)
            and isinstance(message.get('content'), str)
            and 'Earlier conversation history was compacted' in message['content']
        ]
        self.assertEqual(len(compact_messages), 1)
        transcript_compact_messages = [
            message for message in result.transcript
            if message.get('metadata', {}).get('kind') == 'compact_boundary'
        ]
        self.assertEqual(len(transcript_compact_messages), 0)
        model_compact_messages = [
            message for message in (agent.last_session.messages if agent.last_session else [])
            if message.metadata.get('kind') == 'compact_boundary'
        ]
        self.assertEqual(len(model_compact_messages), 1)
        compact_metadata = model_compact_messages[0].metadata
        self.assertEqual(compact_metadata.get('compaction_depth'), 1)
        self.assertEqual(compact_metadata.get('nested_compaction_count'), 0)
        self.assertIn('preserved_tail_ids', compact_metadata)

    def test_agent_enforces_total_token_budget(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'This would be the answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {
                    'prompt_tokens': 30,
                    'completion_tokens': 12,
                },
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        budget_config=BudgetConfig(max_total_tokens=20),
                    ),
                )
                result = agent.run('Use too many tokens')
        self.assertEqual(result.stop_reason, 'budget_exceeded')
        self.assertIn('token budget', result.final_output)
        self.assertEqual(result.usage.total_tokens, 42)

    def test_agent_rejects_prompt_before_backend_when_preflight_input_budget_is_exceeded(self) -> None:
        snapshot = TokenBudgetSnapshot(
            model='test-model',
            context_window_tokens=1000,
            projected_input_tokens=240,
            message_tokens=220,
            chat_overhead_tokens=20,
            reserved_output_tokens=128,
            reserved_compaction_buffer_tokens=64,
            reserved_schema_tokens=0,
            hard_input_limit_tokens=40,
            soft_input_limit_tokens=0,
            overflow_tokens=200,
            soft_overflow_tokens=240,
            exceeds_hard_limit=True,
            exceeds_soft_limit=True,
            token_counter_backend='heuristic',
            token_counter_source='test',
            token_counter_accurate=False,
        )
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.agent_runtime.calculate_token_budget',
                return_value=snapshot,
            ), patch(
                'src.agent_runtime.LocalCodingAgent._reduce_context_pressure',
                return_value=False,
            ), patch(
                'src.openai_compat.request.urlopen',
                side_effect=AssertionError('backend should not be called'),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='test-model',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        budget_config=BudgetConfig(max_input_tokens=20),
                    ),
                )
                result = agent.run(
                    'This prompt is intentionally much longer than the tiny configured input budget. '
                    * 4
                )
        self.assertEqual(result.stop_reason, 'prompt_too_long')
        self.assertIn('Stopped before the next model call', result.final_output)

    def test_agent_auto_compacts_context_before_next_model_call(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Recovered after compaction.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 11, 'completion_tokens': 4},
            }
        ]

        def fake_budget(
            *,
            session,
            model,
            budget_config,
            output_schema=None,
        ):
            compacted = any(
                message.metadata.get('kind') == 'compact_summary'
                for message in session.messages
            )
            projected = 120 if compacted else 540
            return TokenBudgetSnapshot(
                model=model,
                context_window_tokens=1000,
                projected_input_tokens=projected,
                message_tokens=max(projected - 18, 0),
                chat_overhead_tokens=18,
                reserved_output_tokens=128,
                reserved_compaction_buffer_tokens=64,
                reserved_schema_tokens=0,
                hard_input_limit_tokens=420,
                soft_input_limit_tokens=180,
                overflow_tokens=max(projected - 420, 0),
                soft_overflow_tokens=max(projected - 180, 0),
                exceeds_hard_limit=projected > 420,
                exceeds_soft_limit=projected > 180,
                token_counter_backend='heuristic',
                token_counter_source='test',
                token_counter_accurate=False,
            )

        def fake_compact(agent, custom_instructions=None):
            session = agent.last_session
            assert session is not None
            preserved_tail = [session.messages[-1]]
            boundary = AgentMessage(
                role='user',
                content='<system-reminder>Earlier conversation was compacted.</system-reminder>',
                message_id='compact_boundary',
                metadata={'kind': 'compact_boundary'},
            )
            summary = AgentMessage(
                role='user',
                content='Compacted summary.',
                message_id='compact_summary',
                metadata={'kind': 'compact_summary', 'is_compact_summary': True},
            )
            session.messages = [session.messages[0], boundary, summary] + preserved_tail
            return CompactionResult(
                boundary_message=boundary,
                summary_messages=[summary],
                messages_to_keep=preserved_tail,
                pre_compact_token_count=540,
                post_compact_token_count=120,
                summary_text='Summary',
                usage=UsageStats(input_tokens=7, output_tokens=3),
            )

        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            runtime_config = AgentRuntimeConfig(
                cwd=workspace,
                compact_preserve_messages=1,
            )
            stored = StoredAgentSession(
                session_id='resume_auto_compact',
                model_config=serialize_model_config(
                    ModelConfig(
                        model='test-model',
                        base_url='http://127.0.0.1:8000/v1',
                    )
                ),
                runtime_config=serialize_runtime_config(runtime_config),
                system_prompt_parts=('# System\nYou are helpful.',),
                user_context={},
                system_context={},
                messages=(
                    {
                        'role': 'system',
                        'content': '# System\nYou are helpful.',
                        'message_id': 'system_0',
                        'metadata': {'kind': 'system'},
                    },
                    {
                        'role': 'user',
                        'content': 'First request.',
                        'message_id': 'user_1',
                        'metadata': {'kind': 'user'},
                    },
                    {
                        'role': 'assistant',
                        'content': 'First response.',
                        'message_id': 'assistant_1',
                        'metadata': {'kind': 'assistant'},
                    },
                    {
                        'role': 'user',
                        'content': 'Second request.',
                        'message_id': 'user_2',
                        'metadata': {'kind': 'user'},
                    },
                    {
                        'role': 'assistant',
                        'content': 'Second response.',
                        'message_id': 'assistant_2',
                        'metadata': {'kind': 'assistant'},
                    },
                ),
                turns=2,
                tool_calls=0,
                usage=UsageStats().to_dict(),
                total_cost_usd=0.0,
                file_history=(),
                budget_state={},
                plugin_state={},
                scratchpad_directory=None,
            )
            with patch(
                'src.agent_runtime.calculate_token_budget',
                side_effect=fake_budget,
            ), patch(
                'src.agent_runtime.LocalCodingAgent._reduce_context_pressure',
                return_value=False,
            ), patch(
                'src.agent_runtime.compact_conversation',
                side_effect=fake_compact,
            ), patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='test-model',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=runtime_config,
                )
                result = agent.resume('Continue after compaction', stored)
        self.assertEqual(result.final_output, 'Recovered after compaction.')
        self.assertTrue(
            any(event.get('type') == 'auto_compact_summary' for event in result.events)
        )
        self.assertGreaterEqual(result.usage.total_tokens, 25)

    def test_agent_continues_when_model_response_is_truncated(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Part 1 ',
                        },
                        'finish_reason': 'length',
                    }
                ],
                'usage': {'prompt_tokens': 10, 'completion_tokens': 4},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Part 2',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Give me a long answer')
        self.assertEqual(result.final_output, 'Part 1 Part 2')
        continuation_events = [
            event for event in result.events if event.get('type') == 'continuation_request'
        ]
        self.assertEqual(len(continuation_events), 1)
        second_request_messages = recorded_payloads[1]['messages']
        assert isinstance(second_request_messages, list)
        self.assertTrue(
            any(
                isinstance(message, dict)
                and 'Continue exactly where you left off' in str(message.get('content', ''))
                for message in second_request_messages
            )
        )

    def test_agent_stops_when_truncated_response_is_empty(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': '',
                        },
                        'finish_reason': 'length',
                    }
                ],
                'usage': {'prompt_tokens': 10, 'completion_tokens': 0},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Give me a long answer')

        self.assertEqual(len(recorded_payloads), 1)
        self.assertEqual(result.stop_reason, 'empty_truncated_response')
        self.assertIn('空的截断响应', result.final_output)
        self.assertFalse(
            any(event.get('type') == 'continuation_request' for event in result.events)
        )
        self.assertEqual(result.transcript[-1]['stop_reason'], 'empty_truncated_response')

    def test_agent_limits_automatic_truncated_response_continuations(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': f'Part {index} ',
                        },
                        'finish_reason': 'length',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            }
            for index in range(MAX_AUTO_CONTINUATIONS + 1)
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Give me a very long answer')

        continuation_events = [
            event for event in result.events if event.get('type') == 'continuation_request'
        ]
        self.assertEqual(len(recorded_payloads), MAX_AUTO_CONTINUATIONS + 1)
        self.assertEqual(len(continuation_events), MAX_AUTO_CONTINUATIONS)
        self.assertEqual(result.stop_reason, 'length')
        self.assertIn(f'Part {MAX_AUTO_CONTINUATIONS}', result.final_output)
        self.assertIn('自动续写上限', result.final_output)
        self.assertTrue(
            any(event.get('type') == 'continuation_limit_reached' for event in result.events)
        )

    def test_agent_short_circuits_duplicate_skill_invocation_in_one_run(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will load the verification skill.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'Skill',
                                        'arguments': '{"skill": "verify"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I accidentally load it again.',
                            'tool_calls': [
                                {
                                    'id': 'call_2',
                                    'type': 'function',
                                    'function': {
                                        'name': 'Skill',
                                        'arguments': '{"skill": "verify"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Done.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 1},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_urlopen_side_effect(responses),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use verify skill')

        self.assertEqual(result.final_output, 'Done.')
        self.assertEqual(result.tool_calls, 2)
        tool_messages = [
            message for message in result.transcript if message.get('role') == 'tool'
        ]
        self.assertEqual(len(tool_messages), 2)
        self.assertEqual(tool_messages[0].get('metadata', {}).get('action'), 'skill')
        self.assertEqual(
            tool_messages[1].get('metadata', {}).get('action'),
            'skill_duplicate',
        )
        duplicate_payload = json.loads(str(tool_messages[1].get('content', '{}')))
        self.assertIn('已在本轮激活', duplicate_payload.get('content', ''))

    def test_agent_stops_repeated_duplicate_skill_loop(self) -> None:
        skill_call = {
            'type': 'function',
            'function': {
                'name': 'Skill',
                'arguments': '{"skill": "verify"}',
            },
        }
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will load the verification skill.',
                            'tool_calls': [{**skill_call, 'id': 'call_1'}],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I accidentally load it again.',
                            'tool_calls': [{**skill_call, 'id': 'call_2'}],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I am still stuck loading it again.',
                            'tool_calls': [{**skill_call, 'id': 'call_3'}],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace, max_turns=50),
                )
                result = agent.run('Use verify skill')

        self.assertEqual(len(recorded_payloads), 3)
        self.assertEqual(result.stop_reason, 'duplicate_skill_loop')
        self.assertIn('连续重复调用 Skill', result.final_output)
        self.assertEqual(result.tool_calls, 3)
        tool_messages = [
            message for message in result.transcript if message.get('role') == 'tool'
        ]
        self.assertEqual(tool_messages[-1].get('metadata', {}).get('stop_agent'), True)
        self.assertEqual(tool_messages[-1].get('metadata', {}).get('duplicate_count'), 2)
        self.assertTrue(
            any(event.get('type') == 'tool_requested_stop' for event in result.events)
        )

    def test_agent_records_file_history_for_write_tool(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Creating the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'write_file',
                                        'arguments': '{"path": "out.txt", "content": "hi"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Done.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                result = agent.run('Create out.txt')
                stored = load_agent_session(result.session_id or '', directory=session_dir)
        self.assertEqual(len(result.file_history), 1)
        self.assertEqual(result.file_history[0]['path'], 'out.txt')
        self.assertEqual(result.file_history[0]['history_kind'], 'file_change')
        self.assertIn('history_entry_id', result.file_history[0])
        self.assertIn('after_snapshot_id', result.file_history[0])
        self.assertEqual(stored.file_history[0]['action'], 'write_file')
        self.assertEqual(stored.file_history[0]['after_snapshot_id'], result.file_history[0]['after_snapshot_id'])

    def test_agent_streams_write_file_tool_output(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Creating the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'write_file',
                                        'arguments': '{"path": "out.txt", "content": "hello"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Write streamed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                result = agent.run('Create out.txt')
        self.assertEqual(result.final_output, 'Write streamed.')
        tool_delta_events = [
            event for event in result.events
            if event.get('type') == 'tool_delta' and event.get('tool_name') == 'write_file'
        ]
        self.assertGreaterEqual(len(tool_delta_events), 1)
        tool_message = next(
            message for message in result.transcript
            if message.get('role') == 'tool'
        )
        metadata = tool_message.get('metadata', {})
        self.assertEqual(metadata.get('streamed'), True)
        self.assertEqual(metadata.get('action'), 'write_file')

    def test_agent_streams_bash_tool_output_and_mutates_tool_transcript(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Running a shell command.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'bash',
                                        'arguments': json.dumps(
                                            {
                                                'command': (
                                                    "printf 'alpha\\n'; "
                                                    "sleep 0.05; "
                                                    "printf 'beta\\n' >&2"
                                                )
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Shell command completed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        permissions=AgentPermissions(allow_shell_commands=True),
                    ),
                )
                result = agent.run('Run a shell command')
        self.assertEqual(result.final_output, 'Shell command completed.')
        tool_delta_events = [event for event in result.events if event.get('type') == 'tool_delta']
        self.assertGreaterEqual(len(tool_delta_events), 2)
        joined_delta = ''.join(event.get('delta', '') for event in tool_delta_events)
        self.assertIn('alpha', joined_delta)
        self.assertIn('beta', joined_delta)
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        tool_message = tool_messages[0]
        self.assertIn('exit_code=0', tool_message.get('content', ''))
        metadata = tool_message.get('metadata', {})
        self.assertIn('stream_preview', metadata)
        self.assertIn('alpha', metadata['stream_preview'])
        self.assertIn('beta', metadata['stream_preview'])

    def test_agent_streams_read_file_tool_output_in_chunks(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Reading the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "large.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Read finished.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'large.txt').write_text('alpha\n' * 300, encoding='utf-8')
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Read the large file')
        self.assertEqual(result.final_output, 'Read finished.')
        tool_delta_events = [
            event for event in result.events
            if event.get('type') == 'tool_delta' and event.get('tool_name') == 'read_file'
        ]
        self.assertGreaterEqual(len(tool_delta_events), 2)
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        self.assertEqual(tool_messages[0].get('metadata', {}).get('streamed'), True)
        self.assertIn('stream_preview', tool_messages[0].get('metadata', {}))

    def test_agent_records_tombstone_mutation_history_when_snipping(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Reading the large file first.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "large.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Snip run completed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'large.txt').write_text(('alpha beta gamma\n' * 400), encoding='utf-8')
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        auto_snip_threshold_tokens=120,
                        compact_preserve_messages=0,
                    ),
                )
                result = agent.run('Read the large file and summarize it')
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        metadata = tool_messages[0].get('metadata', {})
        self.assertEqual(tool_messages[0].get('state'), 'tombstoned')
        self.assertEqual(metadata.get('kind'), 'snipped_message')
        self.assertEqual(metadata.get('last_mutation_kind'), 'snip_tombstone')
        self.assertGreaterEqual(metadata.get('mutation_count', 0), 2)
        self.assertTrue(any(entry.get('kind') == 'tool_finalize_replace' for entry in metadata.get('mutations', [])))
        self.assertTrue(any(event.get('type') == 'snip_boundary' for event in result.events))

    def test_resume_injects_file_history_replay_reminder(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Creating the file first.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'write_file',
                                        'arguments': '{"path": "replay.txt", "content": "hello"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Initial write done.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resume acknowledged.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                first_result = agent.run('Create replay.txt')
                stored = load_agent_session(first_result.session_id or '', directory=session_dir)

                resumed_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                resumed_agent.resume('Continue the work', stored)
        resumed_messages = recorded_payloads[-1]['messages']
        assert isinstance(resumed_messages, list)
        replay_messages = [
            message for message in resumed_messages
            if isinstance(message, dict)
            and isinstance(message.get('content'), str)
            and 'Recent file history from this saved session:' in message['content']
        ]
        self.assertEqual(len(replay_messages), 1)
        self.assertIn('path=replay.txt', replay_messages[0]['content'])
        self.assertIn('action=write_file', replay_messages[0]['content'])

    def test_resume_replays_file_history_snapshots(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Editing the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'edit_file',
                                        'arguments': json.dumps(
                                            {
                                                'path': 'draft.txt',
                                                'old_text': 'hello world',
                                                'new_text': 'hello mars',
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Edit completed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resume processed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'draft.txt').write_text('hello world\n', encoding='utf-8')
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                first_result = agent.run('Edit draft.txt')
                stored = load_agent_session(first_result.session_id or '', directory=session_dir)
                resumed_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        permissions=AgentPermissions(allow_file_write=True),
                    ),
                )
                resumed_agent.resume('Continue after the edit', stored)
        resumed_messages = recorded_payloads[-1]['messages']
        assert isinstance(resumed_messages, list)
        replay_messages = [
            message for message in resumed_messages
            if isinstance(message, dict)
            and isinstance(message.get('content'), str)
            and 'Recent file history from this saved session:' in message['content']
        ]
        self.assertEqual(len(replay_messages), 1)
        replay_content = replay_messages[0]['content']
        self.assertIn('Unique changed paths: 1', replay_content)
        self.assertIn('Snapshot ids: 2', replay_content)
        self.assertIn('before_snapshot:', replay_content)
        self.assertIn('after_snapshot:', replay_content)
        self.assertIn('before: hello world', replay_content)
        self.assertIn('after: hello mars', replay_content)
        self.assertIn('result:', replay_content)

    def test_resume_injects_compaction_replay_reminder(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Reading the large file first.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'read_file',
                                        'arguments': '{"path": "large.txt"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Initial compaction run completed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resume after compaction processed.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'large.txt').write_text(('alpha beta gamma\n' * 400), encoding='utf-8')
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        auto_snip_threshold_tokens=120,
                        compact_preserve_messages=0,
                    ),
                )
                first_result = agent.run('Read the large file and summarize it')
                stored = load_agent_session(first_result.session_id or '', directory=session_dir)
                resumed_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        auto_snip_threshold_tokens=120,
                        compact_preserve_messages=0,
                    ),
                )
                resumed_agent.resume('Continue after compaction', stored)
        resumed_messages = recorded_payloads[-1]['messages']
        assert isinstance(resumed_messages, list)
        compaction_messages = [
            message for message in resumed_messages
            if isinstance(message, dict)
            and isinstance(message.get('content'), str)
            and 'This resumed session already contains compacted or snipped history.' in message['content']
        ]
        self.assertEqual(len(compaction_messages), 1)
        self.assertIn('Snipped/tombstoned messages:', compaction_messages[0]['content'])

    def test_agent_can_delegate_to_nested_agent(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating this task.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'prompt': 'Summarize the delegated task.',
                                                'max_turns': 2,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegated summary complete.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent task completed after delegation.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use a delegate agent')
        self.assertEqual(result.final_output, 'Parent task completed after delegation.')
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        self.assertIn('Delegated agent completed the subtask.', tool_messages[0]['content'])
        metadata = tool_messages[0].get('metadata', {})
        self.assertEqual(metadata.get('action'), 'delegate_agent')
        self.assertIsNotNone(metadata.get('child_session_id'))

    def test_agent_can_delegate_multiple_subtasks_with_parent_context(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating multiple subtasks.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'subtasks': [
                                                    {'label': 'scan', 'prompt': 'Scan the project.'},
                                                    {'label': 'summarize', 'prompt': 'Summarize the project.'},
                                                ],
                                                'max_turns': 2,
                                                'include_parent_context': True,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child scan result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child summary result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent completed after multi-delegate.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use multiple delegated subtasks')
        self.assertEqual(result.final_output, 'Parent completed after multi-delegate.')
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        metadata = tool_messages[0].get('metadata', {})
        self.assertEqual(metadata.get('subtask_count'), 2)
        self.assertEqual(len(metadata.get('child_session_ids', [])), 2)
        self.assertEqual(len(metadata.get('child_results', [])), 2)
        self.assertIn('Delegated agent completed 2 sequential subtasks.', tool_messages[0].get('content', ''))
        self.assertTrue(any(event.get('type') == 'delegate_group_result' for event in result.events))
        child_events = [event for event in result.events if event.get('type') == 'delegate_subtask_result']
        self.assertEqual(len(child_events), 2)
        second_child_request = recorded_payloads[2]['messages']
        assert isinstance(second_child_request, list)
        self.assertTrue(
            any(
                isinstance(message, dict)
                and 'Prior delegated subtask summaries:' in str(message.get('content', ''))
                for message in second_child_request
            )
        )

    def test_agent_manager_tracks_delegate_group_membership(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating multiple subtasks.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'subtasks': [
                                                    {'label': 'scan', 'prompt': 'Scan the project.'},
                                                    {'label': 'summarize', 'prompt': 'Summarize the project.'},
                                                ],
                                                'max_turns': 2,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child scan result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child summary result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent completed after multi-delegate.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use multiple delegated subtasks')

        self.assertEqual(result.final_output, 'Parent completed after multi-delegate.')
        self.assertIsNotNone(agent.agent_manager)
        manager = agent.agent_manager
        assert manager is not None
        self.assertEqual(len(manager.groups), 1)
        group = next(iter(manager.groups.values()))
        self.assertEqual(group.completed_children, 2)
        child_records = sorted(
            (
                record for record in manager.completed_records()
                if record.parent_agent_id == agent.managed_agent_id
            ),
            key=lambda record: (record.child_index or 0),
        )
        self.assertEqual(len(child_records), 2)
        self.assertEqual([record.child_index for record in child_records], [1, 2])
        self.assertTrue(all(record.group_id == group.group_id for record in child_records))
        summary = '\n'.join(manager.summary_lines())
        self.assertIn(f'group={group.group_id}', summary)

    def test_agent_can_delegate_into_resumed_child_session(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Seed child result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating into a resumed child session.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': '{}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resumed child result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent completed after resumed child.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                seed_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                seeded = seed_agent.run('Seed the delegated child')
                resumed_child_id = seeded.session_id or ''

                delegate_arguments = json.dumps(
                    {
                        'subtasks': [
                            {
                                'label': 'resume_child',
                                'prompt': 'Continue the delegated child.',
                                'resume_session_id': resumed_child_id,
                                'max_turns': 2,
                            }
                        ],
                        'max_turns': 2,
                    }
                )
                responses[1]['choices'][0]['message']['tool_calls'][0]['function']['arguments'] = delegate_arguments

                parent_agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                result = parent_agent.run('Delegate into the resumed child')

        self.assertEqual(result.final_output, 'Parent completed after resumed child.')
        tool_messages = [message for message in result.transcript if message.get('role') == 'tool']
        self.assertEqual(len(tool_messages), 1)
        metadata = tool_messages[0].get('metadata', {})
        self.assertEqual(metadata.get('resumed_children'), 1)
        child_results = metadata.get('child_results', [])
        self.assertEqual(len(child_results), 1)
        self.assertEqual(child_results[0].get('resume_used'), True)
        self.assertEqual(child_results[0].get('resumed_from_session_id'), resumed_child_id)
        self.assertTrue(
            any(
                event.get('type') == 'delegate_subtask_result'
                and event.get('resume_used')
                for event in result.events
            )
        )
        resumed_child_messages = recorded_payloads[2]['messages']
        assert isinstance(resumed_child_messages, list)
        resumed_contents = [
            message.get('content')
            for message in resumed_child_messages
            if isinstance(message, dict)
        ]
        self.assertIn('Seed the delegated child', resumed_contents)
        self.assertIn('Seed child result.', resumed_contents)
        self.assertIn('Continue the delegated child.', resumed_contents)
        manager = parent_agent.agent_manager
        assert manager is not None
        child_records = [
            record
            for record in manager.completed_records()
            if record.parent_agent_id == parent_agent.managed_agent_id
        ]
        self.assertEqual(len(child_records), 1)
        self.assertEqual(child_records[0].resumed_from_session_id, resumed_child_id)
        summary = '\n'.join(manager.summary_lines())
        self.assertIn(f'resumed_from={resumed_child_id}', summary)

    def test_agent_enforces_reasoning_token_budget(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'This uses too much reasoning.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {
                    'prompt_tokens': 8,
                    'completion_tokens': 4,
                    'completion_tokens_details': {'reasoning_tokens': 9},
                },
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        budget_config=BudgetConfig(max_reasoning_tokens=5),
                    ),
                )
                result = agent.run('Use too much reasoning')
        self.assertEqual(result.stop_reason, 'budget_exceeded')
        self.assertIn('reasoning token budget', result.final_output)
        self.assertEqual(result.usage.reasoning_tokens, 9)

    def test_agent_enforces_tool_call_budget_before_execution(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will create the file.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'write_file',
                                        'arguments': '{"path": "blocked.txt", "content": "nope"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        permissions=AgentPermissions(allow_file_write=True),
                        budget_config=BudgetConfig(max_tool_calls=0),
                    ),
                )
                result = agent.run('Try to create a file')
                self.assertFalse((workspace / 'blocked.txt').exists())
        self.assertEqual(result.stop_reason, 'budget_exceeded')
        self.assertIn('tool-call budget', result.final_output)
        self.assertFalse(any(message.get('role') == 'tool' for message in result.transcript))

    def test_agent_enforces_delegated_task_budget_before_child_agent_runs(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'I will delegate this.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'prompt': 'Do the delegated work.',
                                                'max_turns': 2,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        budget_config=BudgetConfig(max_delegated_tasks=0),
                    ),
                )
                result = agent.run('Try to delegate')
        self.assertEqual(result.stop_reason, 'budget_exceeded')
        self.assertIn('delegated-task budget', result.final_output)
        self.assertFalse(any(message.get('role') == 'tool' for message in result.transcript))

    def test_agent_enforces_cumulative_model_call_budget_across_resume(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'First answer.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Second answer should exceed the model-call budget.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        budget_config=BudgetConfig(max_model_calls=1),
                    ),
                )
                first = agent.run('First prompt')
                stored = load_agent_session(first.session_id or '', directory=session_dir)
                resumed = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                        budget_config=BudgetConfig(max_model_calls=1),
                    ),
                )
                second = resumed.resume('Second prompt', stored)
        self.assertEqual(first.stop_reason, 'stop')
        self.assertEqual(second.stop_reason, 'budget_exceeded')
        self.assertIn('model-call budget', second.final_output)

    def test_plugin_runtime_state_is_restored_on_resume(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Calling the virtual plugin tool.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'demo_virtual',
                                        'arguments': '{"topic": "state"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Initial plugin state stored.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resume consumed plugin runtime state.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            plugin_dir = workspace / 'plugins' / 'demo'
            plugin_dir.mkdir(parents=True)
            (plugin_dir / 'plugin.json').write_text(
                json.dumps(
                    {
                        'name': 'demo-plugin',
                        'hooks': {'beforePrompt': 'Plugin hook before prompt.'},
                        'virtualTools': [
                            {
                                'name': 'demo_virtual',
                                'description': 'Emit plugin state.',
                                'responseTemplate': 'plugin state {topic}',
                            }
                        ],
                    }
                ),
                encoding='utf-8',
            )
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                first = agent.run('Use the plugin virtual tool')
                stored = load_agent_session(first.session_id or '', directory=session_dir)
                resumed = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                second = resumed.resume('Continue after plugin state', stored)
        self.assertEqual(second.final_output, 'Resume consumed plugin runtime state.')
        resumed_messages = recorded_payloads[-1]['messages']
        assert isinstance(resumed_messages, list)
        self.assertTrue(
            any(
                isinstance(message, dict)
                and 'Plugin runtime state:' in str(message.get('content', ''))
                and 'virtual_tool_results=1' in str(message.get('content', ''))
                for message in resumed_messages
            )
        )

    def test_plugin_lifecycle_hooks_are_persisted_and_reapplied_on_resume(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Initial lifecycle run stored.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Resume observed plugin lifecycle hooks.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 2},
            },
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            session_dir = workspace / '.port_sessions' / 'agent'
            plugin_dir = workspace / 'plugins' / 'demo'
            plugin_dir.mkdir(parents=True)
            (plugin_dir / 'plugin.json').write_text(
                json.dumps(
                    {
                        'name': 'demo-plugin',
                        'hooks': {
                            'onResume': 'Reapply the saved plugin lifecycle state on resume.',
                            'beforePersist': 'Persist plugin lifecycle markers into the saved session.',
                        },
                    }
                ),
                encoding='utf-8',
            )
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                first = agent.run('Store the plugin lifecycle session')
                stored = load_agent_session(first.session_id or '', directory=session_dir)
                resumed = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        session_directory=session_dir,
                    ),
                )
                second = resumed.resume('Continue after lifecycle persistence', stored)
        self.assertEqual(second.final_output, 'Resume observed plugin lifecycle hooks.')
        self.assertTrue(
            any(
                message.get('metadata', {}).get('kind') == 'plugin_persist'
                and 'Persist plugin lifecycle markers into the saved session.' in str(message.get('content', ''))
                for message in stored.messages
            )
        )
        resumed_messages = recorded_payloads[-1]['messages']
        assert isinstance(resumed_messages, list)
        self.assertTrue(
            any(
                isinstance(message, dict)
                and 'Plugin resume hooks:' in str(message.get('content', ''))
                and 'Reapply the saved plugin lifecycle state on resume.' in str(message.get('content', ''))
                for message in resumed_messages
            )
        )
        self.assertTrue(any(event.get('type') == 'plugin_before_persist' for event in first.events))

    def test_agent_can_delegate_with_dependency_aware_subtasks(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating dependency-aware subtasks.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'subtasks': [
                                                    {
                                                        'label': 'summarize',
                                                        'prompt': 'Summarize the project.',
                                                        'depends_on': ['scan'],
                                                    },
                                                    {
                                                        'label': 'scan',
                                                        'prompt': 'Scan the project.',
                                                    },
                                                ],
                                                'max_turns': 2,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child scan result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent completed after dependency-aware delegation.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use dependency-aware delegated subtasks')
        self.assertEqual(result.final_output, 'Parent completed after dependency-aware delegation.')
        child_events = [event for event in result.events if event.get('type') == 'delegate_subtask_result']
        self.assertEqual(len(child_events), 2)
        self.assertEqual(child_events[0].get('stop_reason'), 'pending_dependency')
        self.assertEqual(child_events[1].get('stop_reason'), 'stop')

    def test_agent_can_delegate_with_topological_batches(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Delegating batched subtasks.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'delegate_agent',
                                        'arguments': json.dumps(
                                            {
                                                'subtasks': [
                                                    {
                                                        'label': 'summarize',
                                                        'prompt': 'Summarize the project.',
                                                        'depends_on': ['scan'],
                                                    },
                                                    {
                                                        'label': 'scan',
                                                        'prompt': 'Scan the project.',
                                                    },
                                                ],
                                                'strategy': 'topological',
                                                'max_turns': 2,
                                            }
                                        ),
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child scan result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Child summary result.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 5, 'completion_tokens': 2},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Parent completed after topological delegation.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 2},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Use topological delegated subtasks')
        self.assertEqual(result.final_output, 'Parent completed after topological delegation.')
        child_events = [event for event in result.events if event.get('type') == 'delegate_subtask_result']
        self.assertEqual([event.get('label') for event in child_events], ['scan', 'summarize'])
        self.assertEqual([event.get('batch_index') for event in child_events], [1, 2])
        batch_events = [event for event in result.events if event.get('type') == 'delegate_batch_result']
        self.assertEqual(len(batch_events), 2)
        self.assertEqual(batch_events[0].get('status'), 'completed')
        delegate_entry = next(
            entry for entry in result.file_history if entry.get('action') == 'delegate_agent'
        )
        self.assertEqual(delegate_entry.get('delegate_batch_count'), 2)
        self.assertEqual(delegate_entry.get('dependency_skips'), 0)
        tool_message = next(
            message
            for message in result.transcript
            if message.get('role') == 'tool'
            and message.get('metadata', {}).get('action') == 'delegate_agent'
        )
        self.assertEqual(tool_message['metadata'].get('strategy'), 'topological')

    def test_agent_sends_response_schema_when_configured(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': '{"status":"ok"}',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 4},
            }
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            with patch(
                'src.openai_compat.request.urlopen',
                side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
            ):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(
                        cwd=workspace,
                        output_schema=OutputSchemaConfig(
                            name='status_response',
                            schema={
                                'type': 'object',
                                'properties': {'status': {'type': 'string'}},
                                'required': ['status'],
                            },
                            strict=True,
                        ),
                    ),
                )
                agent.run('Return a JSON status payload')
        self.assertEqual(
            recorded_payloads[0]['response_format'],
            {
                'type': 'json_schema',
                'json_schema': {
                    'name': 'status_response',
                    'schema': {
                        'type': 'object',
                        'properties': {'status': {'type': 'string'}},
                        'required': ['status'],
                    },
                    'strict': True,
                },
            },
        )

    def test_hook_policy_before_prompt_injection_and_after_turn_event(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Completed under policy guidance.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
            }
        ]
        recorded_payloads: list[dict[str, object]] = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / '.claw-policy.json').write_text(
                (
                    '{"trusted": false, '
                    '"managedSettings": {"reviewMode": "strict"}, '
                    '"safeEnv": ["HOOK_SAFE_TOKEN"], '
                    '"hooks": {"beforePrompt": ["Respect workspace policy."], '
                    '"afterTurn": ["Persist the policy decision."]}}'
                ),
                encoding='utf-8',
            )
            with patch.dict('os.environ', {'HOOK_SAFE_TOKEN': 'demo-secret'}, clear=False):
                with patch(
                    'src.openai_compat.request.urlopen',
                    side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
                ):
                    agent = LocalCodingAgent(
                        model_config=ModelConfig(
                            model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                            base_url='http://127.0.0.1:8000/v1',
                        ),
                        runtime_config=AgentRuntimeConfig(cwd=workspace),
                    )
                    result = agent.run('Inspect the repository.')
        payload_text = json.dumps(recorded_payloads[0])
        self.assertIn('Workspace hook/policy guidance', payload_text)
        self.assertIn('Respect workspace policy.', payload_text)
        self.assertIn('Trust mode: untrusted', payload_text)
        self.assertEqual(agent.tool_context.extra_env.get('HOOK_SAFE_TOKEN'), 'demo-secret')
        self.assertIn('reviewMode=strict', agent.render_trust_report())
        hook_events = [event for event in result.events if event.get('type') == 'hook_policy_after_turn']
        self.assertEqual(len(hook_events), 1)
        self.assertEqual(hook_events[0].get('message'), 'Persist the policy decision.')

    def test_hook_policy_blocks_tool_and_tracks_permission_denial(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Trying bash first.',
                            'tool_calls': [
                                {
                                    'id': 'call_1',
                                    'type': 'function',
                                    'function': {
                                        'name': 'bash',
                                        'arguments': '{"command": "pwd"}',
                                    },
                                }
                            ],
                        },
                        'finish_reason': 'tool_calls',
                    }
                ],
                'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
            },
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'Bash was blocked by workspace policy.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 7, 'completion_tokens': 3},
            },
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / '.claw-policy.json').write_text(
                '{"denyTools": ["bash"]}',
                encoding='utf-8',
            )
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Run pwd')
        self.assertEqual(result.final_output, 'Bash was blocked by workspace policy.')
        event_types = [event.get('type') for event in result.events]
        self.assertIn('hook_policy_tool_block', event_types)
        self.assertIn('tool_permission_denial', event_types)
        tool_message = next(
            message
            for message in result.transcript
            if message.get('role') == 'tool'
        )
        self.assertTrue(tool_message['metadata'].get('hook_policy_blocked'))
        self.assertEqual(tool_message['metadata'].get('error_kind'), 'permission_denied')

    def test_hook_policy_budget_override_applies_when_runtime_budget_is_unset(self) -> None:
        responses = [
            {
                'choices': [
                    {
                        'message': {
                            'role': 'assistant',
                            'content': 'One model call happened.',
                        },
                        'finish_reason': 'stop',
                    }
                ],
                'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
            }
        ]
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / '.claw-policy.json').write_text(
                '{"budget": {"max_model_calls": 0}}',
                encoding='utf-8',
            )
            with patch('src.openai_compat.request.urlopen', side_effect=make_urlopen_side_effect(responses)):
                agent = LocalCodingAgent(
                    model_config=ModelConfig(
                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
                        base_url='http://127.0.0.1:8000/v1',
                    ),
                    runtime_config=AgentRuntimeConfig(cwd=workspace),
                )
                result = agent.run('Hello')
        self.assertEqual(result.stop_reason, 'budget_exceeded')
        self.assertIn('model-call budget was exceeded', result.final_output)