"""File editing, search, and cell timing are exercised through the daemon's real Python tool."""

import json
import unittest

from harness import exclusive, Albedo, Provider, Reply, cell_result, python, text


class PythonToolsTests(unittest.TestCase):
    def setUp(self):
        self.arguments = None
        self.tool = "python"

        def script(request):
            if request["messages"][-1].get("role") == "user":
                if self.arguments is not None:
                    return Reply(
                        "python", tool_name=self.tool, tool_arguments=self.arguments
                    )
                return python(self.code)
            return text("done")

        self.provider = Provider(script)
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider)
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def run_cell(self, code, session):
        """The result of one prompt's cell, the latest in the session."""
        self.code = code
        self.app.prompt(session, "use the Python tools").close()
        self.app.idle(session)
        results = [
            json.loads(part["value"])
            for entry in self.app.history(session)["items"]
            if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python"
            for part in entry["content"]
            if part["kind"] == "json" and part["field"] == "result"
        ]
        self.assertEqual(results[-1]["status"], "ok", results[-1])
        return results[-1]

    def test_large_file_edit_keeps_live_result_inline_without_snapshots(self):
        original = "".join(
            f"line {index:04d}: existing source text\n" for index in range(1300)
        )
        (self.app.workspace / "source.txt").write_text(original)
        session = self.app.session()
        before = self.app.stream_page(session)
        self.run_cell(
            "files.edit('source.txt', 'line 0650:', 'changed 0650:')", session
        )
        live = next(
            event["data"]
            for event in self.app.stream_page(session, before)["events"]
            if event["type"] == "tool"
        )
        self.assertTrue(live["content_complete"], live)
        self.assertIn("changed 0650:", live["trace"]["changes"][0]["diff"])
        result = self.run_cell(
            "previous = (await cells.list(limit=2))[-1]\n"
            "print(json.dumps(await cells.trace(previous['id'])))",
            session,
        )
        change = json.loads(result["output"])["changes"][0]
        self.assertNotIn("old_text", change)
        self.assertNotIn("new_text", change)

    def test_interrupted_cell_reports_failure_live_and_in_history(self):
        self.assert_cell_failure({"code": "raise KeyboardInterrupt()"})

    def test_deadline_cell_reports_failure_live_and_in_history(self):
        self.assert_cell_failure(
            {"code": "await asyncio.Event().wait()", "timeout_ms": 25}
        )

    def assert_cell_failure(self, arguments):
        session = self.app.session()
        self.arguments = arguments
        before = self.app.stream_page(session)
        self.app.prompt(session, "run interrupted cell").close()
        self.app.idle(session)
        self.assert_failed_tool(session, before)

    def assert_failed_tool(self, session, before):
        live = next(
            event["data"]
            for event in self.app.stream_page(session, before)["events"]
            if event["type"] == "tool"
        )
        history = next(
            entry
            for entry in self.app.history(session)["items"]
            if entry["kind"] == "tool_result"
        )
        self.assertEqual(json.loads(live["result"])["status"], "interrupted")
        self.assertEqual(live["status"], "failed")
        self.assertEqual(history["tool"]["status"], "failed")

    def execute(self, code):
        return self.run_cell(code, self.app.session())["output"]

    @exclusive
    def test_host_replies_preserve_escaped_call_ids_and_argument_errors(self):
        output = self.execute(
            "kernel = sys.modules['albedo_cells']\n"
            "key = 'host reply \" \\\\ λ\\n'\n"
            "pending = asyncio.get_running_loop().create_future()\n"
            "kernel.PENDING[key] = pending\n"
            "try:\n"
            "    kernel.send({'type': 'call', 'id': key, 'method': 'mail.submit', 'args': []})\n"
            "    print(json.dumps(await asyncio.wait_for(pending, 5)))\n"
            "finally:\n    kernel.PENDING.pop(key, None)"
        )
        self.assertEqual(
            json.loads(output),
            {
                "ok": False,
                "code": "invalid",
                "message": "mail.submit(to, body) takes two strings",
            },
        )

    @exclusive
    def test_shadowing_injected_tools_is_refused_while_locals_and_stdlib_recovery_work(
        self,
    ):
        (self.app.workspace / "note.txt").write_text("recovered\n")
        session = self.app.session()
        for bad in (
            "ran_before = True\nfiles = 1",
            "del files",
            "def overwrite():\n    global output\n    output = 1",
        ):
            self.call({"code": bad}, session)
            failed = self.sent()
            self.assertEqual(failed["status"], "error")
            self.assertIn("kernel tool", failed["output"])
        output = self.run_cell(
            "assert 'ran_before' not in globals()\n"
            "try:\n"
            "    globals()['files'] = 1\n"
            "except PermissionError as err:\n"
            "    assert 'kernel tool' in str(err)\n"
            "else:\n"
            "    raise AssertionError('runtime globals() overwrite was not refused')\n"
            "from albedo import files\n"
            "def local_scope(files):\n"
            "    return [output for output in (files,)]\n"
            "assert local_scope('ok') == ['ok']\n"
            "json = 'shadowed'\n"
            "assert json == 'shadowed'\n"
            "del json\n"
            "assert json.dumps(1) == '1'\n"
            "import builtins\nassert not hasattr(builtins, 'files')\n"
            "print(files.read('note.txt'))",
            session,
        )["output"]
        self.assertIn("recovered", output)

    def test_ambiguous_edit_requires_a_hint_and_preserves_other_matches(self):
        note = self.app.workspace / "note.txt"
        note.write_text("same\nother\nsame\n")
        output = self.execute(
            "try:\n    files.edit('note.txt', 'same', 'changed')\n"
            "except Exception as exc:\n    print('ambiguous:', exc)\n"
            "files.edit('note.txt', 'same', 'changed', line_hint=3)\n"
            "print(files.read('note.txt'))"
        )
        self.assertEqual(note.read_text(), "same\nother\nchanged\n")
        self.assertIn("ambiguous:", output)
        self.assertIn("same", output)
        self.assertIn("changed", output)

    def test_a_list_of_edits_applies_all_or_none(self):
        note = self.app.workspace / "note.txt"
        note.write_text("alpha\nbeta\ngamma\n")
        output = self.execute(
            "print(files.edit('note.txt', [('alpha', 'one'), ('gamma', 'three')]))\n"
            "try:\n"
            "    files.edit('note.txt', [('one', 'uno'), ('absent', 'x')])\n"
            "except ValueError as error:\n"
            "    print(str(error).splitlines()[0])"
        )
        self.assertEqual(note.read_text(), "one\nbeta\nthree\n")
        lines = output.splitlines()
        self.assertIn("2 replacements", lines[0])
        self.assertIn(
            "edit 2 of 2 failed, so none were made: string not found", lines[1]
        )

    def test_text_returns_plain_output_and_raises_with_the_last_lines(self):
        program = (
            "import sys; print('\\x1b[32mgreen\\x1b[0m'); "
            "print('10%\\r100%'); sys.exit(int(sys.argv[1]))"
        )
        output = self.execute(
            f"program = {program!r}\n"
            "print(repr(await run(sys.executable, '-c', program, '0').text()))\n"
            "try:\n"
            "    await run(sys.executable, '-c', program, '3').text()\n"
            "except RuntimeError as error:\n"
            "    print(type(error).__name__, repr(str(error).splitlines()[0]))\n"
            "    print(repr(error.output))\n"
            "print(repr(await run(sys.executable, '-c', program, '3').text(check=False)))\n"
            # `await` takes the slice with it; the slice applies to the text.
            "print(repr(await run(sys.executable, '-c', program, '0').text()[-5:]))\n"
            "job = run(sys.executable, '-c', program, '3')\n"
            "print(repr(await parallel(job.text(check=False)[:5])))"
        )
        lines = output.splitlines()
        self.assertEqual(lines[0], repr("green\n100%\n"))
        self.assertEqual(lines[1].split()[0], "JobFailed")
        self.assertIn("exited 3", lines[1])
        self.assertEqual(lines[2], repr("green\n100%\n"))
        self.assertEqual(lines[3], repr("green\n100%\n"))
        self.assertEqual(lines[4], repr("100%\n"))
        self.assertEqual(lines[5], repr(["green"]))

    def sent(self):
        """The latest python result as the provider received it."""
        return next(
            cell_result(message["content"])
            for record in reversed(self.provider.requests)
            for message in reversed(record["request"]["messages"])
            if message.get("role") == "tool"
        )

    def test_results_reach_the_model_as_a_json_line_then_the_output(self):
        session = self.app.session()
        self.run_cell("print('one\\n\\t\"two\"')\n41 + 1", session)
        ok = self.sent()
        self.assertEqual(ok["output"], 'one\n\t"two"\n')
        self.assertEqual((ok["status"], ok["value"]), ("ok", "42"))
        self.assertNotIn("cell_id", ok)
        self.assertNotIn("truncated", ok)
        # An awaiting cell without a final expression has no value, and a
        # regex in a plain string runs without a SyntaxWarning riding along.
        self.run_cell(
            "for _ in range(1):\n    await asyncio.sleep(0)\nre.findall('\\d', 'a1')",
            session,
        )
        quiet = self.sent()
        self.assertEqual((quiet["output"], quiet["value"]), ("", "['1']"))
        self.run_cell("for _ in range(1):\n    await asyncio.sleep(0)", session)
        self.assertNotIn("value", self.sent())
        self.run_cell("print('x' * 40000)", session)
        big = self.sent()
        self.assertTrue(big["truncated"])
        self.assertLess(len(big["output"]), 33 * 1024)
        self.assertIn("[output was 40001 bytes", big["output"])
        # A failure names the cell's own frames as <cell>, not the kernel's.
        code = "\n".join(f"x{i} = {i}" for i in range(11)) + "\nprint(x11)"
        self.call({"code": code}, session)
        failed = self.sent()
        self.assertEqual(failed["status"], "error")
        self.assertIn('File "<cell>", line 12', failed["output"])
        self.assertIn("NameError", failed["output"])
        self.assertNotIn("albedo_cells.py", failed["output"])
        self.assertIn(
            "12 lines; fix it in place with await cells.run(", failed["repair"]
        )
        self.assertNotIn("cell_id", failed)

    def test_wait_for_returns_the_line_and_leaves_the_job_running(self):
        server = (
            "import sys, time; print('starting', flush=True); time.sleep(0.4); "
            "print('ready on 4321', flush=True); time.sleep(60)"
        )
        output = self.execute(
            "import re\n"
            f"server = run(sys.executable, '-c', {server!r})\n"
            "print(repr(await server.wait_for('ready', timeout=20)))\n"
            "print(repr(await server.wait_for(re.compile(r'on \\d+'), timeout=1)))\n"
            "print(server.exit_code)\n"
            "try:\n"
            "    await server.wait_for('never', timeout=0.5)\n"
            "except TimeoutError as error:\n"
            "    print('timeout', 'still running' in str(error))\n"
            "await server.stop()\n"
            "quick = run(sys.executable, '-c', 'print(1)')\n"
            "try:\n"
            "    await quick.wait_for('ready', timeout=20)\n"
            "except RuntimeError as error:\n"
            "    print(type(error).__name__, 'without printing' in str(error))"
        )
        self.assertEqual(
            output.splitlines(),
            [
                repr("ready on 4321"),
                repr("ready on 4321"),
                "None",
                "timeout True",
                "JobFailed True",
            ],
        )

    def test_chdir_moves_run_and_files(self):
        (self.app.workspace / "sub").mkdir()
        (self.app.workspace / "sub" / "note.txt").write_text("inside\n")
        output = self.execute(
            "os.chdir('sub')\n"
            "print(await run(sys.executable, '-c', 'import os; print(os.getcwd())').text())\n"
            "print(files.read('note.txt'))"
        )
        self.assertIn(str((self.app.workspace / "sub").resolve()), output)
        self.assertIn("inside", output)

    def test_an_edit_miss_says_when_only_whitespace_differs(self):
        (self.app.workspace / "note.txt").write_text("if ok:\n    run()\n")
        output = self.execute(
            "for old in ('if ok:\\n\\trun()', 'absent'):\n"
            "    try:\n        files.edit('note.txt', old, 'x')\n"
            "    except ValueError as error:\n        print(str(error).splitlines()[0])"
        )
        lines = output.splitlines()
        self.assertIn("whitespace differs", lines[0])
        self.assertNotIn("whitespace differs", lines[1])

    def test_find_searches_workspace_and_returns_context(self):
        (self.app.workspace / "note.txt").write_text("before\nneedle\nafter\n")
        output = self.execute(
            "rows = await files.find('needle', '.', context=1)\n"
            "print(rows)\nprint(files.read('note.txt', start_line=2, end_line=2))"
        )
        self.assertIn("note.txt", output)
        self.assertIn("needle", output)
        self.assertIn("before", output)
        self.assertIn("after", output)

    def test_an_empty_find_says_what_was_searched_and_skipped(self):
        (self.app.workspace / ".ignore").write_text("ignored.txt\n")
        (self.app.workspace / "ignored.txt").write_text("needle\n")
        (self.app.workspace / ".dot.txt").write_text("needle\n")
        (self.app.workspace / "plain.txt").write_text("nothing\n")
        output = self.execute(
            "print(await files.find('needle', '.'))\n"
            "print(await files.find('needle', '.', hidden=True, ignored=True))"
        )
        self.assertIn("no matches for 'needle'", output)
        self.assertIn("1 files searched", output)
        self.assertIn("skipped hidden files, .gitignore/.ignore rules", output)
        self.assertIn("ignored.txt", output)
        self.assertIn("rerun with hidden=True and/or ignored=True", output)
        self.assertEqual(output.count("no matches"), 1, output)

    def test_find_in_a_missing_root_raises_instead_of_returning_nothing(self):
        output = self.execute(
            "try:\n    await files.find('needle', 'nowhere')\n"
            "except FileNotFoundError as exc:\n    print('missing:', exc)"
        )
        self.assertIn("missing: nowhere not found", output)

    def test_a_file_result_is_awaitable_and_offers_content_and_text(self):
        (self.app.workspace / "note.txt").write_text("hello\n")
        output = self.execute(
            "text = await files.read('note.txt')\n"
            "print('awaited:', text.strip())\n"
            "print('content:', (await files.read('note.txt')).content.strip())\n"
            "print('text:', files.read('note.txt').text.strip())"
        )
        self.assertIn("awaited: 1 | hello", output)
        self.assertIn("content: 1 | hello", output)
        self.assertIn("text: 1 | hello", output)

    def test_paths_searches_a_list_of_directories_with_a_list_of_globs(self):
        for directory, name in [("a", "one.md"), ("b", "two.txt"), ("c", "three.md")]:
            (self.app.workspace / directory).mkdir()
            (self.app.workspace / directory / name).write_text("x")
        output = self.execute(
            "rows = await files.paths(None, ['a', 'b'], glob=['*.md', '*.txt'])\n"
            "print(sorted(str(row) for row in rows))"
        )
        self.assertIn("one.md", output)
        self.assertIn("two.txt", output)
        self.assertNotIn("three.md", output)

    def test_paths_globs_a_path_from_the_workspace_root(self):
        (self.app.workspace / "src" / "deep").mkdir(parents=True)
        (self.app.workspace / "src" / "deep" / "stream_test.go").write_text("x")
        output = self.execute(
            "print([str(row) for row in await files.paths('src/*/*_test.go')])\n"
            "print([str(row) for row in await files.paths('./src/*/*_test.go')])"
        )
        self.assertEqual(output.count("['src/deep/stream_test.go']"), 2, output)

    @exclusive
    def test_paths_globs_with_color_enabled_in_ripgrep_config(self):
        config = self.app.workspace / "ripgrep.conf"
        config.write_text("--pretty\n")
        self.app.restart(
            prepare=lambda app: app.daemon.env.update(RIPGREP_CONFIG_PATH=str(config))
        )
        (self.app.workspace / "src" / "deep").mkdir(parents=True)
        (self.app.workspace / "src" / "deep" / "stream_test.go").write_text("x")
        output = self.execute(
            "print([str(row) for row in await files.paths('src/*/*_test.go')])\n"
            "print([str(row) for row in await files.paths('./src/*/*_test.go')])"
        )
        self.assertEqual(output.count("['src/deep/stream_test.go']"), 2, output)

    def test_file_reads_decode_complete_lines_across_chunk_boundaries(self):
        fixtures = [
            (b"", []),
            (b"\n\n", ["", ""]),
            (b"a\r\r\n", ["a", ""]),
            (b"a\n", ["a"]),
            (b"a\nlast", ["a", "last"]),
            (
                "a\nb\rc\r\nd\ve\ff\x1cg\x1dh\x1ei\x85j\u2028k\u2029last".encode(),
                list("abcdefghijk") + ["last"],
            ),
            (b"a\xffb\xe2\x82", ["a\ufffdb\ufffd"]),
        ]
        for padding in (65_534, 65_535, 65_536):
            prefix = b"x" * padding
            for separator in (
                b"\r\n",
                "\x85".encode(),
                "\u2028".encode(),
                "\u2029".encode(),
            ):
                fixtures.append((prefix + separator + b"tail", ["x" * padding, "tail"]))
            fixtures.append((prefix + "é\n尾".encode(), ["x" * padding + "é", "尾"]))
            fixtures.append(
                (
                    prefix + b"\xe2\x82\nend\xe2\x82",
                    ["x" * padding + "\ufffd", "end\ufffd"],
                )
            )
        for index, (data, _lines) in enumerate(fixtures):
            (self.app.workspace / f"decode-{index}.txt").write_bytes(data)
        # Keep large expected lines inside the kernel, out of the tool output.
        small_expected = [lines for _data, lines in fixtures[:7]]
        output = self.execute(
            f"expected = {small_expected!r}\n"
            "for padding in (65534, 65535, 65536):\n"
            "    expected.extend([['x' * padding, 'tail']] * 4)\n"
            "    expected.append(['x' * padding + 'é', '尾'])\n"
            "    expected.append(['x' * padding + '\ufffd', 'end\ufffd'])\n"
            "for index, lines in enumerate(expected):\n"
            "    result = files.read(f'decode-{index}.txt', max_chars=200000)\n"
            "    numbered = '\\n'.join(f'{number:>6} | {line}' for number, line in enumerate(lines, 1))\n"
            "    if lines:\n"
            "        assert result == numbered, (index, len(result), len(numbered))\n"
            "    else:\n"
            "        assert ' | ' not in result\n"
            "        assert all(detail in result for detail in ('decode-0.txt', '0 lines', 'line 1'))\n"
            "print('decoded complete lines')"
        )
        self.assertEqual(output.strip(), "decoded complete lines")

    def test_file_reads_enforce_budgets_and_character_limit_precedence(self):
        (self.app.workspace / "budget.txt").write_text("a\nbb\nc\nd\n")
        (self.app.workspace / "numbered.txt").write_text("\n" * 999_999 + "x\nz")
        output = self.execute(
            "first = '     1 | a'\n"
            "second = '     2 | bb'\n"
            "both = first + '\\n' + second\n"
            "assert files.read('budget.txt', end_line=2, max_chars=23) == both\n"
            "assert files.read('budget.txt', end_line=1, max_chars=11) == first\n"
            "rejected = files.read('budget.txt', max_chars=10)\n"
            "assert ' | ' not in rejected\n"
            "assert all(detail in rejected for detail in ('line 1', '10 characters', 'max_chars=10', 'start_line=1', 'end_line=1', 'max_chars=11'))\n"
            "for budget, shown, unshown, resume in ((22, [first], 'lines 2-4', 2), (23, [first, second], 'lines 3-4', 3)):\n"
            "    result = files.read('budget.txt', max_chars=budget).splitlines()\n"
            "    assert result[:-1] == shown\n"
            "    assert all(detail in result[-1] for detail in (f'max_chars={budget}', unshown, f'start_line={resume}'))\n"
            "result = files.read('budget.txt', limit=2, max_chars=23).splitlines()\n"
            "assert result[:-1] == [first, second]\n"
            "assert all(detail in result[-1] for detail in ('limit=2', '2 more through line 4', 'start_line=3'))\n"
            "result = files.read('budget.txt', limit=3, max_chars=23).splitlines()\n"
            "assert result[:-1] == [first, second]\n"
            "assert all(detail in result[-1] for detail in ('max_chars=23', 'lines 3-3', 'start_line=3'))\n"
            "assert 'limit=' not in result[-1]\n"
            "assert files.read('numbered.txt', start_line=1000000, end_line=1000000, max_chars=12) == '1000000 | x'\n"
            "rejected = files.read('numbered.txt', start_line=1000000, end_line=1000000, max_chars=11)\n"
            "assert ' | ' not in rejected\n"
            "assert all(detail in rejected for detail in ('line 1000000', '11 characters', 'max_chars=11', 'start_line=1000000', 'end_line=1000000', 'max_chars=12'))\n"
            "print('budgets and numbering checked')"
        )
        self.assertEqual(output.strip(), "budgets and numbering checked")

    def test_file_reads_count_requested_windows_and_eof_exactly(self):
        (self.app.workspace / "window.txt").write_text("a\nb\nc\nd")
        pattern = "a\r\nb\rc\nd\ve\ff\x1cg\x1dh\x1ei\x85j\u2028k\u2029last\n".encode()
        (self.app.workspace / "bulk.txt").write_bytes(
            b"head\r\n" + pattern * 10_000 + b"tail\xe2\x82"
        )
        output = self.execute(
            "assert files.read('window.txt', start_line=2, end_line=3) == '     2 | b\\n     3 | c'\n"
            "assert files.read('window.txt', start_line=3, end_line=99) == '     3 | c\\n     4 | d'\n"
            "for options, row, remaining, last, resume in (({'start_line': 2}, '     2 | b', 2, 4, 3), ({'start_line': 2, 'end_line': 3}, '     2 | b', 1, 3, 3), ({'start_line': 3, 'end_line': 99}, '     3 | c', 1, 4, 4)):\n"
            "    result = files.read('window.txt', limit=1, **options).splitlines()\n"
            "    assert result[:-1] == [row]\n"
            "    assert all(detail in result[-1] for detail in ('limit=1', f'{remaining} more through line {last}', f'start_line={resume}'))\n"
            "assert files.read('window.txt', start_line=4, limit=1) == '     4 | d'\n"
            "empty = files.read('window.txt', start_line=5, end_line=99)\n"
            "assert all(detail in empty for detail in ('window.txt', '4 lines', 'line 5'))\n"
            "assert ' | ' not in empty\n"
            "result = files.read('bulk.txt', start_line=2, limit=1).splitlines()\n"
            "assert result[:-1] == ['     2 | a']\n"
            "assert all(detail in result[-1] for detail in ('limit=1', '120000 more through line 120002', 'start_line=3'))\n"
            "result = files.read('bulk.txt', start_line=2, end_line=100000, limit=1).splitlines()\n"
            "assert result[:-1] == ['     2 | a']\n"
            "assert all(detail in result[-1] for detail in ('limit=1', '99998 more through line 100000', 'start_line=3'))\n"
            "result = files.read('bulk.txt', start_line=120001, limit=1).splitlines()\n"
            "assert result[:-1] == ['120001 | last']\n"
            "assert all(detail in result[-1] for detail in ('limit=1', '1 more through line 120002', 'start_line=120002'))\n"
            "assert files.read('bulk.txt', start_line=120001, end_line=120001) == '120001 | last'\n"
            "assert files.read('bulk.txt', start_line=120002) == '120002 | tail\\ufffd'\n"
            "print('window counts checked')"
        )
        self.assertEqual(output.strip(), "window counts checked")

    def test_file_reads_count_giant_lines_without_shortening_them(self):
        (self.app.workspace / "giant.txt").write_text(
            "head\n" + "x" * 1_048_576 + "\ntail\n"
        )
        (self.app.workspace / "unicode.txt").write_text("é" * 110_000)
        output = self.execute(
            "assert files.read('giant.txt', start_line=3) == '     3 | tail'\n"
            "assert files.read('giant.txt', end_line=1) == '     1 | head'\n"
            "rejected = files.read('giant.txt', start_line=2, end_line=2)\n"
            "assert ' | ' not in rejected\n"
            "assert all(detail in rejected for detail in ('line 2', '1048585 characters', 'max_chars=16000', '200000'))\n"
            "assert 'max_chars=1048586' not in rejected\n"
            "result = files.read('giant.txt', max_chars=32).splitlines()\n"
            "assert result[:-1] == ['     1 | head']\n"
            "assert all(detail in result[-1] for detail in ('max_chars=32', 'lines 2-3', 'start_line=2'))\n"
            "result = files.read('giant.txt', limit=1).splitlines()\n"
            "assert result[:-1] == ['     1 | head']\n"
            "assert all(detail in result[-1] for detail in ('limit=1', '2 more through line 3', 'start_line=2'))\n"
            "for budget in (16000, 110009):\n"
            "    rejected = files.read('unicode.txt', max_chars=budget)\n"
            "    assert ' | ' not in rejected\n"
            "    assert all(detail in rejected for detail in ('line 1', '110009 characters', f'max_chars={budget}', 'start_line=1', 'end_line=1', 'max_chars=110010'))\n"
            "assert files.read('unicode.txt', max_chars=110010) == '     1 | ' + 'é' * 110000\n"
            "print('giant lines counted and returned intact')"
        )
        self.assertEqual(output.strip(), "giant lines counted and returned intact")

    def test_file_reads_validate_ranges_before_opening_and_report_path_errors(self):
        output = self.execute(
            "for options in ({'start_line': 0}, {'start_line': 2, 'end_line': 1}, {'limit': 0}, {'max_chars': 0}, {'max_chars': 200001}):\n"
            "    try:\n"
            "        files.read('absent.txt', **options)\n"
            "    except ValueError:\n"
            "        pass\n"
            "    else:\n"
            "        raise AssertionError(options)\n"
            "try:\n"
            "    files.read('absent.txt')\n"
            "except FileNotFoundError as error:\n"
            "    assert 'absent.txt' in str(error)\n"
            "else:\n"
            "    raise AssertionError('missing file accepted')\n"
            "try:\n"
            "    files.read('.')\n"
            "except IsADirectoryError as error:\n"
            "    assert \"files.ls('.')\" in str(error)\n"
            "else:\n"
            "    raise AssertionError('directory accepted')\n"
            "print('range and path errors preserved')"
        )
        self.assertEqual(output.strip(), "range and path errors preserved")

    def test_cells_get_the_api_a_model_reaches_for(self):
        output = self.execute(
            "job = run('echo', 'hello')\n"
            "await job\n"
            "assert (await job.tail()).strip() == 'hello'\n"
            "assert (await job.head(lines=1)).strip() == 'hello'\n"
            "assert output.read(job.id, 0, 3) == 'hel'\n"
            "try:\n"
            "    output.read(job.id, limit=10**7)\n"
            "except ValueError as error:\n"
            "    assert 'offset=' in str(error), error\n"
            "else:\n"
            "    raise AssertionError('over-cap limit shortened silently')\n"
            "from albedo import files as imported\n"
            "assert imported is files\n"
            "assert re.fullmatch('a+', 'aa') and Path('.').is_dir()\n"
            "assert hashlib.md5(b'').hexdigest() and json.dumps(os.sep)\n"
            "try:\n"
            "    run('true', timeout=120000)\n"
            "except ValueError as error:\n"
            "    assert 'seconds' in str(error) and 'milliseconds' in str(error), error\n"
            "try:\n"
            "    time.sleep(2)\n"
            "except PermissionError as error:\n"
            "    assert 'run()' in str(error) and 'asyncio.sleep' in str(error), error\n"
            "time.sleep(0.01)\n"
            "await asyncio.sleep(0.01)\n"
            "print('api shapes ok')"
        )
        self.assertEqual(output.strip(), "api shapes ok")

    def test_a_cell_reports_how_long_it_ran(self):
        session = self.app.session()
        slept = self.run_cell("import asyncio\nawait asyncio.sleep(0.3)", session)
        self.assertGreaterEqual(slept["duration"], 0.3)
        self.assertLess(slept["duration"], 5)
        # The journal keeps it, so cells.info reports it after the fact.
        info = self.run_cell(
            f"print((await cells.info({slept['cell_id']!r}))['duration'])", session
        )
        self.assertEqual(float(info["output"]), slept["duration"])

    def test_parallel_awaits_handles_that_do_not_hash(self):
        # A remote reference compares by value and is unhashable, so
        # asyncio.gather's own dict of arguments refused it outright.
        output = self.execute(
            "class Ref:\n"
            "    __hash__ = None\n"
            "    def __init__(self, value):\n"
            "        self.value = value\n"
            "    async def _wait(self):\n"
            "        return self.value\n"
            "    def __await__(self):\n"
            "        return self._wait().__await__()\n"
            "print(await parallel(Ref(1), [Ref(2), Ref(3)]))"
        )
        self.assertEqual(output.strip(), "[1, 2, 3]")

    def call(self, arguments, session, tool="python"):
        """The result of one call to `tool` made with exactly these arguments."""
        self.arguments, self.tool = arguments, tool
        self.app.prompt(session, "call the Python tool").close()
        self.app.idle(session)
        self.arguments, self.tool = None, "python"
        return [
            json.loads(part["value"])
            for entry in self.app.history(session)["items"]
            if entry["kind"] == "tool_result" and entry["tool"]["name"] == tool
            for part in entry["content"]
            if part["kind"] == "json" and part["field"] == "result"
        ][-1]

    def test_a_call_naming_a_binding_runs_as_its_cell(self):
        session = self.app.session()
        (self.app.workspace / "note.txt").write_text("from the binding\n")
        read = self.call({"path": "note.txt"}, session, tool="files.read")
        self.assertEqual(read["status"], "ok", read)
        self.assertIn("from the binding", read["value"])
        missing = self.call({}, session, tool="nothing.here")
        self.assertIn("NameError", missing["output"], missing)

    def test_timeout_is_optional(self):
        result = self.call({"code": "print('no timeout')"}, self.app.session())
        self.assertEqual(result["status"], "ok", result)
        self.assertEqual(result["output"].strip(), "no timeout")

    def test_an_unusable_timeout_keeps_the_code_for_a_rerun(self):
        session = self.app.session()
        rejected = self.call(
            {"code": "open('made.txt', 'w').write('hi')", "timeout_ms": "soon"},
            session,
        )
        self.assertIn("timeout_ms", rejected["error"])
        self.assertFalse((self.app.workspace / "made.txt").exists())
        self.run_cell(f"await cells.run({rejected['cell_id']!r})", session)
        self.assertEqual((self.app.workspace / "made.txt").read_text(), "hi")

    def test_a_failed_cell_names_its_repair(self):
        session = self.app.session()
        code = "\n".join(f"x{i} = {i}" for i in range(11)) + "\nprint(x11)"
        long = self.call({"code": code}, session)
        repair = f"await cells.run({long['cell_id']!r}, replacements=[(old, new)], allow_partial=True)"
        self.assertEqual(
            f"12 lines; fix it in place with {repair}, which reruns the lines before the failure too",
            long["repair"],
        )
        fixed = self.run_cell(
            f"await cells.run({long['cell_id']!r}, replacements=[('print(x11)', 'print(x10)')], allow_partial=True)",
            session,
        )
        self.assertEqual(fixed["output"].splitlines()[-1], "10")
        # A short cell is cheaper to resend than to repair.
        short = self.call({"code": "print(missing)"}, session)
        self.assertNotIn("repair", short)
        unstarted = self.call({"code": code.replace("print(x11)", "print((")}, session)
        self.assertEqual(
            f"12 lines; fix it in place with await cells.run({unstarted['cell_id']!r}, replacements=[(old, new)])",
            unstarted["repair"],
        )
        # A repair that fails again names the copy it saved, not the cell
        # that ran it or the original.
        nested = self.call(
            {"code": f"await cells.run({long['cell_id']!r}, allow_partial=True)"},
            session,
        )
        self.assertNotIn(nested["cell_id"], nested["repair"])
        self.assertNotIn(long["cell_id"], nested["repair"])
        self.assertIn(
            "replacements=[(old, new)], allow_partial=True)", nested["repair"]
        )

    def test_workspace_code_cannot_reword_the_repair_note(self):
        session = self.app.session()
        (self.app.workspace / "_fmt.py").write_text(
            "import sys\n"
            "for module in list(sys.modules.values()):\n"
            "    capture = getattr(module, 'Capture', None)\n"
            "    if isinstance(capture, type) and hasattr(capture, 'write'):\n"
            "        write = capture.write\n"
            "        capture.write = lambda self, text, write=write: write(\n"
            "            self, text.replace('fix it in place', 'FORGED fix it in place')\n"
            "        )\n"
            "        break\n"
        )
        self.call({"code": "import _fmt"}, session)
        code = "\n".join(f"x{i} = {i}" for i in range(11)) + "\nprint(missing)"
        failed = self.call({"code": code}, session)
        self.assertNotIn("FORGED", json.dumps(failed))
        self.assertIn(
            f"fix it in place with await cells.run({failed['cell_id']!r}",
            failed["repair"],
        )

    def test_cell_title_on_settled_entry_and_long_cell_progress(self):
        session = self.app.session()
        # Code longer than 512 codepoints so the title comment scrolls off the preview tail.
        code = "# compute aggregate report\n" + "x = 1\n" * 150 + "print(x)\n"
        self.code = code

        before = self.app.stream_page(session)
        self.app.prompt(session, "run the long python cell").close()
        self.app.idle(session)

        # 1. Verify progress events carried the title even when preview offset > 0
        events = self.app.stream_page(session, before)["events"]
        progress_events = [
            e["data"]["progress"]
            for e in events
            if e.get("type") == "tool_progress"
            and e.get("data", {}).get("progress") is not None
        ]
        titled_events = [
            p for p in progress_events if p.get("title") == "compute aggregate report"
        ]
        self.assertTrue(titled_events, "expected progress event with cell title")
        # Ensure at least one event had dropped the title from preview text due to the 512 cap
        long_previews = [
            p
            for p in titled_events
            if p.get("preview", {}).get("offset_scalars", 0) > 0
        ]
        self.assertTrue(
            long_previews, "expected tail preview where title was dropped from preview"
        )
        self.assertNotIn(
            "compute aggregate report", long_previews[-1]["preview"]["text"]
        )

        # 2. Verify settled cell in cells.list() has the extracted title comment exactly without '#'
        cell_info = self.run_cell(
            "listed = await cells.list(limit=5)\n"
            "matching = [c['first_line'] for c in listed if c['first_line'] == 'compute aggregate report']\n"
            "assert len(matching) == 1, [c['first_line'] for c in listed]\n"
            "print('EXACT MATCH')",
            session,
        )
        self.assertEqual(cell_info["output"].strip(), "EXACT MATCH")

    def test_e2e_settled_titles(self):
        session = self.app.session()

        # (a) python cell with title comment -> tool.title equals title without '#'
        self.code = "# summarize repository structure\nprint('hello')\n"
        self.app.prompt(session, "run titled cell").close()
        self.app.idle(session)
        items = self.app.history(session)["items"]
        tool_call_a = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertEqual(
            tool_call_a["tool"].get("title"), "summarize repository structure"
        )

        # (b) python cell with no comment -> tool.title is first code line
        self.code = "y = 42\nprint(y)\n"
        self.app.prompt(session, "run untitled cell").close()
        self.app.idle(session)
        items = self.app.history(session)["items"]
        tool_call_b = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertEqual(tool_call_b["tool"].get("title"), "y = 42")

        # (c) arguments exceeding 64k -> content_complete false, arguments null on stream event, title carried on stream & history
        self.code = "# massive payload cell\nx = '" + ("a" * 70000) + "'\n"
        before = self.app.stream_page(session)
        self.app.prompt(session, "run large cell").close()
        self.app.idle(session)
        events = self.app.stream_page(session, before)["events"]
        tool_events = [e for e in events if e.get("type") == "tool"]
        self.assertTrue(tool_events)
        stream_tool = tool_events[-1]["data"]
        self.assertFalse(stream_tool["content_complete"])
        self.assertIsNone(stream_tool["arguments"])
        self.assertEqual(stream_tool.get("title"), "massive payload cell")
        items = self.app.history(session)["items"]
        tool_call_c = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertEqual(tool_call_c["tool"].get("title"), "massive payload cell")

        # (d) non-python tool -> title null
        (self.app.workspace / "test.txt").write_text("hello world")
        self.call({"path": "test.txt"}, session, tool="files.read")
        items = self.app.history(session)["items"]
        tool_call_d = [
            e
            for e in items
            if e["kind"] == "tool_call" and e["tool"]["name"] == "files.read"
        ][-1]
        self.assertIsNone(tool_call_d["tool"].get("title"))

        # (e) key order other than code-first still resolves title
        self.arguments = {"timeout_ms": 1000, "code": "# title with key order\nx = 1\n"}
        self.app.prompt(session, "run key order cell").close()
        self.app.idle(session)
        self.arguments = None
        items = self.app.history(session)["items"]
        tool_call_e = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertEqual(tool_call_e["tool"].get("title"), "title with key order")

        # (f) 512-scalar bound: 1000 leading blank lines then title -> null title
        self.code = ("\n" * 1000) + "# late title\nprint('late')\n"
        self.app.prompt(session, "run late cell").close()
        self.app.idle(session)
        items = self.app.history(session)["items"]
        tool_call_f = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertIsNone(tool_call_f["tool"].get("title"))

        # (g) within 512-scalar bound: 100 leading blank lines then title -> title found
        self.code = ("\n" * 100) + "# early title\nprint('early')\n"
        self.app.prompt(session, "run early cell").close()
        self.app.idle(session)
        items = self.app.history(session)["items"]
        tool_call_g = [e for e in items if e["kind"] == "tool_call"][-1]
        self.assertEqual(tool_call_g["tool"].get("title"), "early title")

    def test_cell_subprocess_and_blocking_sleep_translate_to_run(self):
        session = self.app.session()
        sub = self.call(
            {
                "code": "import subprocess\nsubprocess.check_output(['sh', '-c', 'git status | head -n 5'])"
            },
            session,
        )
        self.assertEqual(sub["status"], "error")
        self.assertIn("use run() instead of subprocess:", sub["output"])
        self.assertIn("run('git', 'status')", sub["output"])
        slp = self.call({"code": "import time\ntime.sleep(1.5)"}, session)
        self.assertEqual(slp["status"], "error")
        self.assertIn(
            "time.sleep(1.5) blocks the kernel; use run() for programs or `await asyncio.sleep(1.5)`.",
            slp["output"],
        )