"""Human shell execution stays durable, bounded, and separate from model turns."""

from concurrent.futures import ThreadPoolExecutor
import json
import os
from pathlib import Path
import shlex
import shutil
import sqlite3
import sys
import threading
import unittest
import urllib.error

from provider_events import responses_events

from harness import (
    Albedo,
    Provider,
    Reply,
    exclusive,
    error,
    operation_id,
    python,
    release_fifo,
    text,
    wait_until,
)


def python_command(source):
    return shlex.join([sys.executable, "-c", source])


def started_response(value, started, release):
    """Flush response.created, then hold the rest of a successful stream."""
    reply = text(value)

    def events():
        stream = responses_events(reply, "", "fixture-call", 20)
        yield next(stream)
        started.set()
        if not release.wait(20):
            raise AssertionError("fixture never released the started response")
        yield from stream

    return Reply("text", events=events())


def process_running(pid):
    try:
        os.kill(pid, 0)
        stat = Path(f"/proc/{pid}/stat")
        return not stat.exists() or stat.read_text().rsplit(")", 1)[1].split()[0] != "Z"
    except (ProcessLookupError, FileNotFoundError):
        return False


@exclusive
class ShellCommandTests(unittest.TestCase):
    # Shell startup reads HOME and SHELL: every daemon needs a private environment.
    def setUp(self):
        self.script = lambda _: text("reported")
        self.provider = Provider(lambda request: self.script(request))
        self.addCleanup(self.provider.close)

        def prepare(app):
            app.daemon.env["HOME"] = str(app.home)
            app.daemon.env["SHELL"] = shutil.which("bash")
            for key in ("BASH_ENV", "ENV", "ZDOTDIR"):
                app.daemon.env.pop(key, None)
            (app.home / ".bash_profile").write_text(
                "export BANG_LOGIN_PROFILE=loaded\n"
            )

        self.app = Albedo(self.provider, protocol="responses", prepare=prepare)
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.session = self.app.session()

    def url(self, command_id):
        return f"/sessions/{self.session}/shell-commands/{command_id}"

    def put(self, command_id, command):
        with self.app.api(
            self.url(command_id), {"command": command}, method="PUT"
        ) as response:
            return json.load(response)["shell_command"]

    def read(self, command_id):
        with self.app.api(self.url(command_id)) as response:
            return json.load(response)["shell_command"]

    def done(self, command_id, *, committed=True):
        observed = []

        def ready():
            observed[:] = [self.read(command_id)]
            command = observed[0]
            return command["status"] != "running" and (
                not committed or command["history_seq"] is not None
            )

        wait_until(ready, 15, f"shell command {command_id} did not finish")
        return observed[0]

    def shell_history(self):
        return [
            item
            for item in self.app.history(self.session)["items"]
            if item["kind"] == "shell_command"
        ]

    def output_page(self, command_id, offset=0, limit=65536):
        with self.app.api(
            self.url(command_id) + f"/output?offset={offset}&limit={limit}"
        ) as response:
            return json.load(response)

    def test_login_shell_cwd_stdin_and_exit_status_without_model_generation(self):
        command_id = operation_id()
        command = (
            'printf "%s\\n" "$PWD" "$BANG_LOGIN_PROFILE"; '
            + python_command(
                'import os,sys; print("tty",os.isatty(0)); '
                'print("stdin",len(sys.stdin.buffer.read()))'
            )
            + '; printf "stderr-marker\\n" >&2; exit 7'
        )
        self.put(command_id, command)
        result = self.done(command_id)
        self.assertEqual(result["status"], "completed")
        self.assertEqual(result["exit_code"], 7)
        self.assertIn(str(self.app.workspace), result["output"])
        self.assertIn("loaded", result["output"])
        self.assertIn("tty False", result["output"])
        self.assertIn("stdin 0", result["output"])
        self.assertIn("stderr-marker", result["output"])
        self.assertEqual(self.provider.requests, [])
        self.assertEqual(len(self.shell_history()), 1)
        self.assertEqual(self.app.idle(self.session)["status"]["phase"], "idle")

    def test_result_is_in_next_model_request_without_fabricated_tool_calls(self):
        command_id = operation_id()
        command = "printf 'user-shell-foobar\\n'"
        self.put(command_id, command)
        result = self.done(command_id)
        history = self.shell_history()
        self.assertEqual(len(history), 1)
        self.assertEqual(history[0]["position"], result["history_seq"])
        metadata = next(
            part["value"]
            for part in history[0]["content"]
            if part["kind"] == "json" and part["field"] == "shell_command"
        )
        self.assertEqual(metadata["id"], command_id)
        self.assertEqual(metadata["command"], command)
        self.assertEqual(self.provider.requests, [])
        self.app.prompt(self.session, "What did the user command print?").close()
        self.app.idle(self.session)
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertIn("user-shell-foobar", sent)
        self.assertIn("printf", sent)
        self.assertNotIn("function_call_output", sent)

    def test_model_reads_output_beyond_its_context_excerpt_through_read_only_binding(
        self,
    ):
        command_id = operation_id()
        self.put(
            command_id,
            python_command(
                'import sys; sys.stdout.write("h"*65536+"read-through-daemon")'
            ),
        )
        self.done(command_id)

        def script(request):
            if any(
                item.get("type") == "function_call_output" for item in request["input"]
            ):
                return text("read")
            return python(
                f"page = await output.read_shell({command_id!r}, offset=65536); "
                "assert page.next_offset is None; print(page.text)"
            )

        self.script = script
        self.app.prompt(self.session, "Read the rest of that command output.").close()
        self.app.idle(self.session)
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertIn("read-through-daemon", sent)
        self.assertEqual(len(self.provider.requests), 2)

    def test_same_identity_retries_execute_once_and_changed_intent_conflicts(self):
        command_id = operation_id()
        command = python_command(
            'from pathlib import Path; p=Path("effect"); p.open("a").write("x")'
        )
        with ThreadPoolExecutor(max_workers=6) as executor:
            results = list(
                executor.map(lambda _: self.put(command_id, command), range(6))
            )
        self.done(command_id)
        self.assertTrue(all(item["id"] == command_id for item in results))
        self.assertEqual((self.app.workspace / "effect").read_text(), "x")
        with self.assertRaises(urllib.error.HTTPError) as failure:
            self.put(command_id, command + "; printf changed")
        self.assertEqual(failure.exception.code, 409)
        self.assertEqual(len(self.shell_history()), 1)

    def test_completion_during_final_provider_response_gets_a_followup_step(self):
        entered, release = threading.Event(), threading.Event()
        self.addCleanup(release.set)

        def script(_request):
            entered.set()
            if not release.wait(15):
                raise AssertionError("fixture never released model turn")
            return text("current-agent-turn")

        self.script = script
        self.app.prompt(self.session, "Hold this response.").close()
        self.assertTrue(entered.wait(5))
        try:
            command_id = operation_id()
            self.put(command_id, "printf 'parallel-shell-result\\n'")
            result = self.done(command_id, committed=False)
            self.assertIsNone(result["history_seq"])
            self.assertEqual(result["delivery"], "pending")
            self.assertEqual(self.shell_history(), [])
            with self.app.api(f"/sessions/{self.session}?tail=0") as response:
                snapshot = json.load(response)
            self.assertIn(
                command_id, [item["id"] for item in snapshot["shell_commands"]]
            )
            reset = self.app.stream_page(self.session, tail=0)
            self.assertIn(command_id, json.dumps(reset))
            self.assertEqual(len(self.provider.requests), 1)
        finally:
            release.set()
        self.app.idle(self.session)
        result = self.done(command_id)
        items = self.app.history(self.session)["items"]
        assistant = next(item for item in items if item["kind"] == "assistant")
        self.assertLess(assistant["position"], result["history_seq"])
        self.assertEqual(result["delivery"], "delivered")
        self.assertEqual(len(self.provider.requests), 2)
        followup = json.dumps(self.provider.requests[1]["request"]["input"])
        self.assertIn("parallel-shell-result", followup)

    def test_second_running_command_is_refused_and_stop_reaps_descendants(self):
        command_id = operation_id()
        source = (
            "import os,subprocess,sys,time; from pathlib import Path; "
            'child=subprocess.Popen([sys.executable,"-c","import time;time.sleep(120)"]); '
            'Path("pids").write_text(f"{os.getpid()} {child.pid}"); '
            'print("started",flush=True); time.sleep(120)'
        )
        self.put(command_id, python_command(source))
        marker = self.app.workspace / "pids"
        wait_until(marker.exists, 10, "shell process did not start")
        pids = list(map(int, marker.read_text().split()))
        with self.assertRaises(urllib.error.HTTPError) as failure:
            self.put(operation_id(), "printf should-not-run")
        self.assertEqual(failure.exception.code, 409)
        with self.app.api(self.url(command_id) + "/stop", {}) as response:
            json.load(response)
        result = self.done(command_id)
        self.assertEqual(result["status"], "cancelled")
        wait_until(
            lambda: not any(map(process_running, pids)),
            10,
            "shell descendants survived stop",
        )
        self.assertEqual(self.provider.requests, [])

    def test_utf8_output_pages_are_complete_and_terminal_controls_are_inert(self):
        command_id = operation_id()
        visible = "α🌙z" * 12
        source = f'import sys; sys.stdout.write({visible!r}); sys.stdout.flush(); sys.stdout.buffer.write(b"\\x1b]52;c;Zm9v\\x07")'
        self.put(command_id, python_command(source))
        result = self.done(command_id)
        pieces, offset = [], 0
        while True:
            page = self.output_page(command_id, offset, 7)
            pieces.append(page["text"])
            if page["next_offset"] is None:
                break
            self.assertGreater(page["next_offset"], offset)
            offset = page["next_offset"]
        full = "".join(pieces)
        self.assertIn(visible, full)
        self.assertNotIn("\x1b", full)
        self.assertNotIn("\x07", full)
        self.assertEqual(result["output_bytes"], len(full.encode()))
        self.assertFalse(result["truncated"])

    def test_large_capture_is_bounded_but_retained_end_is_reachable(self):
        command_id = operation_id()
        command = python_command('import sys; sys.stdout.write("x"*(17*1024*1024))')
        self.put(command_id, command)
        result = self.done(command_id)
        self.assertTrue(result["truncated"])
        self.assertEqual(result["output_bytes"], 16 * 1024 * 1024)
        self.assertLessEqual(len(result["output"].encode()), 65536)
        tail = self.output_page(command_id, result["output_bytes"] - 32)
        self.assertEqual(tail, {"text": "x" * 32, "next_offset": None})
        self.app.prompt(self.session, "Inspect the command output.").close()
        self.app.idle(self.session)
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertNotIn("x" * 20000, sent)

    def test_fork_preserves_capture_and_rewind_keeps_retry_receipt_without_reexecution(
        self,
    ):
        self.app.prompt(self.session, "before shell").close()
        self.app.idle(self.session)
        first = next(
            item
            for item in self.app.history(self.session)["items"]
            if item["kind"] == "user"
        )
        command_id = operation_id()
        command = "printf x >> fork-effect; printf forked-shell-output"
        self.put(command_id, command)
        self.done(command_id)
        self.app.prompt(self.session, "after shell").close()
        self.app.idle(self.session)
        last = next(
            item
            for item in reversed(self.app.history(self.session)["items"])
            if item["kind"] == "user"
        )
        branch = operation_id()
        with self.app.api(
            f"/sessions/{branch}",
            {
                "kind": "fork",
                "source_session_id": self.session,
                "checkpoint_id": last["checkpoint_id"],
            },
            method="PUT",
            headers={"If-None-Match": "*"},
        ) as response:
            json.load(response)
        copied = [
            item
            for item in self.app.history(branch)["items"]
            if item["kind"] == "shell_command"
        ]
        self.assertEqual(len(copied), 1)
        with self.app.api(
            f"/sessions/{branch}/shell-commands/{command_id}/output"
        ) as response:
            self.assertEqual(json.load(response)["text"], "forked-shell-output")
        self.app.prompt(branch, "What did the shell print?").close()
        self.app.idle(branch)
        self.assertIn(
            "forked-shell-output",
            json.dumps(self.provider.requests[-1]["request"]["input"]),
        )
        with self.app.api(
            f"/sessions/{self.session}/rewind",
            {
                "checkpoint_id": first["checkpoint_id"],
            },
        ) as response:
            json.load(response)
        self.assertEqual(self.shell_history(), [])
        recovered = self.put(command_id, command)
        self.assertEqual(recovered["status"], "completed")
        self.assertIsNone(recovered["history_seq"])
        self.assertEqual(recovered["delivery"], "detached")
        self.assertEqual(self.shell_history(), [])
        self.assertEqual((self.app.workspace / "fork-effect").read_text(), "x")
        self.assertEqual(len(self.provider.requests), 3)

    def test_plain_message_or_assistant_bang_text_is_not_executable(self):
        self.script = lambda _: text("!touch assistant-must-not-run")
        self.app.prompt(self.session, "!touch user-message-must-not-run").close()
        self.app.idle(self.session)
        self.assertFalse((self.app.workspace / "assistant-must-not-run").exists())
        self.assertFalse((self.app.workspace / "user-message-must-not-run").exists())
        self.assertEqual(self.shell_history(), [])
        self.assertEqual(len(self.provider.requests), 1)

    def test_capture_storage_failure_stops_effects_and_records_failed_outcome(self):
        def fail_capture(_app):
            with sqlite3.connect(self.app.home / "albedo.sqlite") as db:
                db.execute(
                    "CREATE TRIGGER bang_shell_capture_failure BEFORE INSERT ON shell_command_output "
                    "BEGIN SELECT RAISE(ABORT, 'simulated capture failure'); END"
                )

        self.app.restart(prepare=fail_capture)
        command_id = operation_id()
        self.put(
            command_id,
            python_command(
                "import os,time; from pathlib import Path; "
                'Path("pid").write_text(str(os.getpid())); print("capture",flush=True); '
                "time.sleep(120)"
            ),
        )
        marker = self.app.workspace / "pid"
        wait_until(marker.exists, 10, "shell process did not start")
        pid = int(marker.read_text())
        result = self.done(command_id)
        self.assertEqual(result["status"], "failed")
        self.assertTrue(result["error"])
        wait_until(
            lambda: not process_running(pid),
            10,
            "command survived capture persistence failure",
        )
        self.assertEqual(self.provider.requests, [])

    def test_session_deletion_reaps_command_and_removes_output(self):
        command_id = operation_id()
        self.put(
            command_id,
            python_command(
                "import os,time; from pathlib import Path; "
                'Path("pid").write_text(str(os.getpid())); print("retained",flush=True); '
                "time.sleep(120)"
            ),
        )
        marker = self.app.workspace / "pid"
        wait_until(marker.exists, 10, "shell process did not start")
        pid = int(marker.read_text())
        resource = f"/sessions/{self.session}?view=configuration"
        with self.app.api(resource) as response:
            validator = response.getheader("ETag")
            response.read()
        with self.app.api(
            resource, method="DELETE", headers={"If-Match": validator}
        ) as response:
            self.assertEqual(json.load(response)["state"], "complete")
        wait_until(
            lambda: not process_running(pid),
            10,
            "shell process survived session deletion",
        )
        with self.assertRaises(urllib.error.HTTPError) as failure:
            self.read(command_id)
        self.assertEqual(failure.exception.code, 404)

    def test_graceful_restart_stops_command_without_rerunning_it(self):
        command_id = operation_id()
        command = python_command(
            "import os,time; from pathlib import Path; "
            'Path("effect").open("a").write("x"); '
            'Path("pid").write_text(str(os.getpid())); time.sleep(120)'
        )
        self.put(command_id, command)
        marker = self.app.workspace / "pid"
        wait_until(marker.exists, 10, "shell process did not start")
        pid = int(marker.read_text())
        self.app.restart()
        recovered = self.done(command_id, committed=False)
        self.assertIn(recovered["status"], ("cancelled", "interrupted"))
        self.app.idle(self.session)  # reconnect materializes the idle actor
        result = self.done(command_id)
        self.assertIn(result["status"], ("cancelled", "interrupted"))
        self.assertEqual(self.put(command_id, command)["status"], result["status"])
        self.assertEqual((self.app.workspace / "effect").read_text(), "x")
        wait_until(
            lambda: not process_running(pid),
            10,
            "shell process survived daemon shutdown",
        )
        self.assertEqual(self.provider.requests, [])

    def test_daemon_crash_interrupts_command_without_replaying_its_effect(self):
        command_id = operation_id()
        command = python_command(
            "import os,time; from pathlib import Path; "
            'Path("effect").open("a").write("x"); '
            'Path("pid").write_text(str(os.getpid())); time.sleep(120)'
        )
        self.put(command_id, command)
        marker = self.app.workspace / "pid"
        wait_until(marker.exists, 10, "shell process did not start")
        pid = int(marker.read_text())
        self.app.restart(crash=True)
        self.assertEqual(
            self.done(command_id, committed=False)["status"], "interrupted"
        )
        self.app.idle(self.session)  # reconnect materializes the idle actor
        result = self.done(command_id)
        self.assertEqual(result["status"], "interrupted")
        self.assertEqual(self.put(command_id, command)["status"], "interrupted")
        self.assertEqual((self.app.workspace / "effect").read_text(), "x")
        wait_until(
            lambda: not process_running(pid), 10, "shell process survived daemon crash"
        )
        self.assertEqual(len(self.shell_history()), 1)
        self.assertEqual(self.provider.requests, [])

    def test_pending_receipt_survives_restart_then_delivers_without_rerunning(self):
        command_id = operation_id()
        command = "printf x >> receipt-effect; printf '%s%s' restart-receipt- 416"
        self.put(command_id, command)
        self.assertEqual(self.done(command_id)["delivery"], "pending")
        self.app.restart(crash=True)
        self.app.idle(self.session)
        self.assertEqual(self.read(command_id)["delivery"], "pending")
        self.assertEqual(self.provider.requests, [])
        self.app.prompt(self.session, "Read the recovered shell receipt.").close()
        self.app.idle(self.session)
        self.assertEqual(self.read(command_id)["delivery"], "delivered")
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertEqual(sent.count("restart-receipt-416"), 1)
        self.app.restart(crash=True)
        self.app.idle(self.session)
        self.assertEqual(self.put(command_id, command)["delivery"], "delivered")
        self.assertEqual((self.app.workspace / "receipt-effect").read_text(), "x")
        self.assertEqual(len(self.shell_history()), 1)
        self.assertEqual(len(self.provider.requests), 1)


class ShellDeliveryTests(unittest.TestCase):
    """Shell receipts join model context at a safe point, not at turn end."""

    url = ShellCommandTests.url
    put = ShellCommandTests.put
    read = ShellCommandTests.read
    done = ShellCommandTests.done
    shell_history = ShellCommandTests.shell_history

    def setUp(self):
        self.script = lambda _: text("done")
        self.provider = Provider(lambda request: self.script(request))
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider, protocol="responses")
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.session = self.app.session()

    def delivery(self, command_id, expected):
        wait_until(
            lambda: self.read(command_id)["delivery"] == expected,
            20,
            f"shell {command_id} was not {expected}",
        )
        return self.read(command_id)

    def gate_cell(self):
        fifo = self.app.workspace / "cell-gate"
        os.mkfifo(fifo)
        self.addCleanup(release_fifo, fifo)
        started = self.app.workspace / "cell-started"
        child = f"open({str(fifo)!r}, 'rb').read(1)"
        code = (
            f"from pathlib import Path\nPath({str(started)!r}).touch()\n"
            f"import sys\nj = run(sys.executable, '-c', {child!r})\n"
            "await j\nprint('cell finished')"
        )
        return fifo, started, code

    def test_completed_shell_is_delivered_after_cell_before_final_answer(self):
        fifo, started, code = self.gate_cell()
        second_request, finish = threading.Event(), threading.Event()
        self.addCleanup(finish.set)
        seen = []

        def script(request):
            seen.append(request)
            if len(seen) == 1:
                return python(code)
            return started_response("answer after shell", second_request, finish)

        self.script = script
        self.app.prompt(self.session, "Start a gated cell.").close()
        wait_until(started.exists, 20, "model cell never started")
        cursor = self.app.stream_page(self.session, tail=0)
        command_id = operation_id()
        self.put(command_id, "printf '%s%s' shell-receipt- 7351; printf '%020000d' 0")
        result = self.done(command_id, committed=False)
        self.assertEqual(result["delivery"], "pending")
        self.assertEqual(len(self.provider.requests), 1)
        with self.app.api(f"/sessions/{self.session}?tail=0") as response:
            snapshot = json.load(response)
        record = next(c for c in snapshot["shell_commands"] if c["id"] == command_id)
        self.assertEqual(record["delivery"], "pending")
        self.app.prompt(self.session, "steering-between-receipts").close()
        later_id = operation_id()
        self.put(later_id, "printf '%s%s' later-fifo-receipt- 481")
        self.assertEqual(self.done(later_id, committed=False)["delivery"], "pending")
        wait_until(lambda: release_fifo(fifo), 20, "cell never opened its gate")
        self.assertTrue(second_request.wait(20), "next model step never started")
        sent = json.dumps(seen[1]["input"])
        self.assertEqual(sent.count("shell-receipt-7351"), 1, sent)
        self.assertEqual(sent.count("steering-between-receipts"), 1, sent)
        self.assertEqual(sent.count("later-fifo-receipt-481"), 1, sent)
        self.assertLess(
            sent.index("shell-receipt-7351"), sent.index("steering-between-receipts")
        )
        self.assertLess(
            sent.index("steering-between-receipts"),
            sent.index("later-fifo-receipt-481"),
        )
        self.assertIn("Human shell command", sent)
        self.assertNotIn("0" * 20000, sent)
        delivered = self.delivery(command_id, "delivered")
        self.assertIsNotNone(delivered["history_seq"])
        events = self.app.stream_page(self.session, after=cursor)
        shell_events = [
            event["data"]["shell_command"]
            for event in events["events"]
            if event.get("type") == "shell_command"
            and event.get("data", {}).get("shell_command", {}).get("id") == command_id
        ]
        self.assertIn("pending", [item["delivery"] for item in shell_events])
        self.assertIn("delivered", [item["delivery"] for item in shell_events])
        finish.set()
        self.app.idle(self.session)
        history = self.app.history(self.session)["items"]
        receipt = next(item for item in history if item["kind"] == "shell_command")
        answer = next(
            item
            for item in history
            if item["kind"] == "assistant" and "answer after shell" in json.dumps(item)
        )
        self.assertLess(receipt["position"], answer["position"])
        self.assertEqual(len(self.shell_history()), 2)
        steering = next(
            item for item in history if "steering-between-receipts" in json.dumps(item)
        )
        later_receipt = next(
            item
            for item in self.shell_history()
            if item["position"] != receipt["position"]
        )
        self.assertLess(receipt["position"], steering["position"])
        self.assertLess(steering["position"], later_receipt["position"])
        self.delivery(later_id, "delivered")
        metadata = next(
            part["value"]
            for part in receipt["content"]
            if part.get("field") == "shell_command"
        )
        self.assertEqual(metadata["delivery"], "delivered")
        self.assertTrue(receipt.get("turn_id"))

    def test_interrupt_keeps_pending_shell_in_history_without_waking_model(self):
        fifo, started, code = self.gate_cell()
        self.script = lambda _: python(code)
        self.app.prompt(self.session, "Start a gated cell to interrupt.").close()
        wait_until(started.exists, 20, "model cell never started")
        command_id = operation_id()
        self.put(command_id, "printf '%s%s' interrupted-receipt- 774")
        self.assertEqual(self.done(command_id, committed=False)["delivery"], "pending")
        with self.app.api(f"/sessions/{self.session}?tail=0") as response:
            snapshot = json.load(response)
        with self.app.api(
            f"/sessions/{self.session}/interrupt",
            {
                "run_id": snapshot["status"]["run_id"],
                "through_input_order": snapshot["input_order"],
            },
        ) as response:
            response.read()
        self.app.idle(self.session)
        result = self.done(command_id)
        self.assertEqual(result["delivery"], "pending")
        self.assertEqual(len(self.shell_history()), 1)
        self.assertEqual(len(self.provider.requests), 1)
        release_fifo(fifo)
        self.assertEqual(len(self.provider.requests), 1)

    def test_fork_preserves_pending_receipt_and_delivers_it_once(self):
        command_id = operation_id()
        self.put(command_id, "printf '%s%s' fork-pending-receipt- 328")
        self.assertEqual(self.done(command_id)["delivery"], "pending")
        self.script = lambda _: error(400, "keep the receipt pending")
        self.app.prompt(self.session, "Create a checkpoint after the receipt.").close()
        self.app.idle(self.session)
        self.assertEqual(self.read(command_id)["delivery"], "pending")
        checkpoint = next(
            item["checkpoint_id"]
            for item in reversed(self.app.history(self.session)["items"])
            if item["kind"] == "user"
        )
        branch = operation_id()
        with self.app.api(
            f"/sessions/{branch}",
            {
                "kind": "fork",
                "source_session_id": self.session,
                "checkpoint_id": checkpoint,
            },
            method="PUT",
            headers={"If-None-Match": "*"},
        ) as response:
            json.load(response)
        with self.app.api(
            f"/sessions/{branch}/shell-commands/{command_id}"
        ) as response:
            self.assertEqual(
                json.load(response)["shell_command"]["delivery"], "pending"
            )
        self.script = lambda _: text("fork accepted receipt")
        self.app.prompt(branch, "Read the forked shell receipt.").close()
        self.app.idle(branch)
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertEqual(sent.count("fork-pending-receipt-328"), 1)
        with self.app.api(
            f"/sessions/{branch}/shell-commands/{command_id}"
        ) as response:
            self.assertEqual(
                json.load(response)["shell_command"]["delivery"], "delivered"
            )
        self.assertEqual(self.read(command_id)["delivery"], "pending")
        branch_history = [
            item
            for item in self.app.history(branch)["items"]
            if item["kind"] == "shell_command"
        ]
        self.assertEqual(len(branch_history), 1)

    def test_idle_receipts_are_seen_once_and_all_are_acknowledged(self):
        commands = []
        for number in range(33):
            command_id = operation_id()
            marker = f"idle-receipt-{number:02d}"
            self.put(command_id, f"printf '%s%s' idle-receipt- {number:02d}")
            result = self.done(command_id)
            self.assertEqual(result["delivery"], "pending")
            commands.append((command_id, marker))
        self.assertEqual(self.provider.requests, [])
        self.app.prompt(self.session, "Read the idle shell receipts.").close()
        self.app.idle(self.session)
        self.assertEqual(len(self.provider.requests), 1)
        sent = json.dumps(self.provider.requests[0]["request"]["input"])
        for command_id, marker in commands:
            self.assertEqual(sent.count(marker), 1, marker)
            self.assertEqual(
                self.delivery(command_id, "delivered")["delivery"], "delivered"
            )
        self.assertEqual(len(self.shell_history()), 33)

    def test_rejected_initial_request_does_not_consume_pending_receipt(self):
        command_id = operation_id()
        self.put(command_id, "printf '%s%s' rejected-receipt- 239")
        self.assertEqual(self.done(command_id)["delivery"], "pending")
        self.script = lambda _: error(400, "fixture deliberately rejected request")
        self.app.prompt(self.session, "Read that receipt.").close()
        self.app.idle(self.session)
        self.assertEqual(self.read(command_id)["delivery"], "pending")
        self.assertEqual(len(self.shell_history()), 1)
        self.script = lambda _: text("accepted")
        self.app.prompt(self.session, "Try the receipt again.").close()
        self.app.idle(self.session)
        sent = json.dumps(self.provider.requests[-1]["request"]["input"])
        self.assertEqual(sent.count("rejected-receipt-239"), 1)
        self.delivery(command_id, "delivered")
        self.assertEqual(len(self.shell_history()), 1)

    def test_shell_finishing_during_request_waits_for_following_safe_point(self):
        fifo, started, code = self.gate_cell()
        entered, release, second = (
            threading.Event(),
            threading.Event(),
            threading.Event(),
        )
        finish = threading.Event()
        self.addCleanup(release.set)
        self.addCleanup(finish.set)
        seen = []

        def script(request):
            seen.append(request)
            if len(seen) == 1:
                entered.set()
                if not release.wait(20):
                    raise AssertionError("first request was not released")
                return python(code)
            return started_response("finished", second, finish)

        self.script = script
        first, later = operation_id(), operation_id()
        self.put(first, "printf '%s%s' first-receipt- 901")
        self.assertEqual(self.done(first)["delivery"], "pending")
        self.app.prompt(self.session, "Read the first command.").close()
        self.assertTrue(entered.wait(20), "provider request did not start")
        self.put(later, "printf '%s%s' later-receipt- 902")
        self.assertEqual(self.done(later, committed=False)["delivery"], "pending")
        self.assertNotIn("later-receipt-902", json.dumps(seen[0]["input"]))
        self.assertEqual(self.read(first)["delivery"], "pending")
        release.set()
        wait_until(started.exists, 20, "first response tool did not start")
        self.delivery(first, "delivered")
        self.assertEqual(self.read(later)["delivery"], "pending")
        wait_until(lambda: release_fifo(fifo), 20, "cell never opened its gate")
        self.assertTrue(second.wait(20), "subsequent request did not start")
        sent = json.dumps(seen[1]["input"])
        self.assertEqual(sent.count("first-receipt-901"), 1)
        self.assertEqual(sent.count("later-receipt-902"), 1)
        self.delivery(first, "delivered")
        self.delivery(later, "delivered")
        finish.set()
        self.app.idle(self.session)