okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

dawnkernel: a job_stop frame lets /jobs stop remote background jobs57a82be2026-10-11

"""Sessions whose kernel lives on another host, reached over ssh.

The daemon's PATH holds a loopback ssh that runs each command locally under
its own "remote" HOME, so remote paths, the staged bundle and the run
directories are all distinct from the daemon's. Hosts named `oldhost`,
`locked` and `gone` stand for a python that is too old, a host that needs a
person to sign in, and one that cannot be reached.
"""

import json
import os
import re
import shlex
import shutil
import signal
import subprocess
import sys
import threading
import time
import unittest
import urllib.error
import urllib.parse
from pathlib import Path

from harness import Albedo, Provider, exclusive, operation_id, python, text

FAKE_SSH = """#!/bin/sh
printf '%s\\n' "$*" >> {log}
query=
while [ $# -gt 0 ]; do
  case "$1" in
    -o) shift 2 ;;
    -G) query=1; shift ;;
    -*) shift ;;
    *) break ;;
  esac
done
target=$1
shift
if [ -n "$query" ]; then
  # ssh -G only reads the config: muxed's names a ControlPath, no other does
  if [ "$target" = muxed ]; then
    echo "controlmaster auto"
    echo "controlpath {mux}"
    echo "controlpersist 1800"
  fi
  exit 0
fi
if [ "$target" = fakehost ] && [ -e {down} ]; then
  echo "ssh: connect to host fakehost port 22: Operation timed out" >&2
  exit 255
fi
case "$target" in
  gone) echo "ssh: connect to host gone port 22: Connection refused" >&2; exit 255 ;;
  locked) echo "locked: Permission denied (publickey)." >&2; exit 255 ;;
  slow-*) sleep 3; echo "ssh: connect to host $target port 22: Connection refused" >&2; exit 255 ;;
esac
case "$*" in
  *.part.*) [ -e {slow} ] && sleep 3 ;;
esac
path={path}
[ "$target" = oldhost ] && path={old}:$path
exec env -i HOME={home} PATH="$path" SHELL={bin}/loginsh /bin/sh -c "$*"
"""

# A login shell whose "profile" greets on stdout and stderr, as many do, and
# sources nothing else: /etc/profile would put the system python ahead of the
# fake host's. Every scenario runs through it, so a greeting must never reach
# a framed stream, nor the daemon's log.
LOGIN_SHELL = """#!/bin/sh
if [ "$1" = -l ]; then
  shift
  echo "welcome to the fake host"
  echo "you have no mail" >&2
fi
exec /bin/sh "$@"
"""

# The daemon's own python3, logging what it runs, so a test can tell that a
# termination ladder never ran on this machine.
LOCAL_PYTHON = """#!/bin/sh
printf '%s\\n' "$*" >> {log}
exec {python} "$@"
"""


def write(path, content):
    path.write_text(content)
    path.chmod(0o755)


def processes(*needles):
    """Pids whose command line holds every needle."""
    listing = subprocess.run(
        ["ps", "-axo", "pid=,command="], capture_output=True, text=True, check=True
    ).stdout
    return [
        int(line.split(None, 1)[0])
        for line in listing.splitlines()
        if all(needle in line for needle in needles)
    ]


def wait_for(predicate, timeout=15, message="condition not met"):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        result = predicate()
        if result:
            return result
        time.sleep(0.1)
    raise AssertionError(message)


# exclusive: sets daemon PATH for loopback SSH and local Python wrappers
@exclusive
class RemoteKernelTests(unittest.TestCase):
    remote_home: Path
    ssh_log: Path
    mux: Path
    python_log: Path
    down: Path
    slow_stage: Path

    def setUp(self):
        self.cells = []

        def script(request):
            if request["input"][-1].get("role") == "user" and self.cells:
                return python(self.cells.pop(0))
            return text("ok")

        def prepare(app):
            root = app.root
            self.remote_home = root / "remote-home"
            self.remote_home.mkdir()
            self.ssh_log = root / "ssh.log"
            self.python_log = root / "local-python.log"
            self.down = root / "fakehost-down"
            self.slow_stage = root / "slow-stage"
            self.mux = root / "cm-muxed"
            real = shutil.which("python3") or sys.executable
            local, remote, old = (
                root / "local-bin",
                root / "remote-bin",
                root / "old-bin",
            )
            for directory in (local, remote, old):
                directory.mkdir()
            write(
                local / "ssh",
                FAKE_SSH.format(
                    log=self.ssh_log,
                    bin=remote,
                    old=old,
                    home=self.remote_home,
                    down=self.down,
                    slow=self.slow_stage,
                    mux=self.mux,
                    path=shlex.quote(f"{remote}{os.pathsep}{app.daemon.env['PATH']}"),
                ),
            )
            write(
                local / "python3", LOCAL_PYTHON.format(log=self.python_log, python=real)
            )
            write(remote / "loginsh", LOGIN_SHELL)
            os.symlink(real, remote / "python3")
            ssh_config = root / "user-home" / ".ssh"
            ssh_config.mkdir()
            (ssh_config / "config").write_text(
                "Host configured-box\n  User someone\nHost *\n  ServerAliveInterval 30\n"
                "Include extra.conf\n"
            )
            (ssh_config / "extra.conf").write_text("Host included-box other-*\n")
            write(old / "python3", "#!/bin/sh\necho 3.9\n")
            app.daemon.env["PATH"] = f"{local}{os.pathsep}{app.daemon.env['PATH']}"

        self.provider = Provider(script)
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider, protocol="responses", prepare=prepare)
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.project = self.remote_home / "proj"
        self.project.mkdir()

    def create(self, workspace):
        session = operation_id()
        with self.app.api(
            f"/sessions/{session}",
            {
                "kind": "new",
                "workspace": workspace,
                "provider_profile": self.app.profile,
            },
            method="PUT",
            headers={"If-None-Match": "*"},
        ) as response:
            return json.load(response)

    def host(self, target):
        query = urllib.parse.urlencode({"target": target})
        with self.app.api("/hosts?" + query) as response:
            [host] = json.load(response)["items"]
        self.assertEqual(host["target"], target)
        return host

    def group(self, workspace):
        query = urllib.parse.urlencode({"workspace": str(workspace)})
        with self.app.api(f"/extensions/links/groups?{query}") as response:
            return json.load(response)

    def merge(self, workspace, other):
        group = self.group(workspace)["configuration_resource"]
        target = self.group(other)["configuration_resource"]
        with self.app.api(
            group["url"],
            {"other_workspace": str(other), "other_etag": target["etag"]},
            headers={"If-Match": group["etag"]},
        ) as response:
            return json.load(response)

    def cell(self, session, code, *, during=None):
        """Run one cell through a model turn and return its tool result."""
        self.cells.append(code)
        self.app.prompt(session, "run it").close()
        if during:
            during()
        self.app.idle(session, timeout=60)
        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"
        ]
        return results[-1]

    def test_a_remote_kernel_runs_in_the_remote_folder_from_the_staged_bundle(self):
        session = self.create(f"fakehost:{self.project}")["id"]
        result = self.cell(
            session,
            "import os, albedo_bundle\n(os.getcwd(), os.environ['HOME'], str(albedo_bundle.ROOT))",
        )
        self.assertEqual(result["status"], "ok", result)
        cwd, home, bundle = eval(result["value"])
        self.assertEqual(os.path.realpath(cwd), os.path.realpath(self.project))
        self.assertEqual(home, str(self.remote_home))
        self.assertTrue(
            bundle.startswith(f"{os.path.realpath(self.remote_home)}/.albedo-remote/")
            or bundle.startswith(f"{self.remote_home}/.albedo-remote/"),
            bundle,
        )
        # The model is told where it runs.
        system = json.dumps(self.provider.requests[0]["request"])
        self.assertIn("execute on fakehost (", system)
        log = (self.app.home / "daemon.log").read_text(errors="replace")
        self.assertNotIn("welcome to the fake host", log)
        self.assertNotIn("you have no mail", log)

    def test_remote_job_output_is_not_a_silently_shortened_patch(self):
        session = self.create(str(self.app.workspace))["id"]
        result = self.cell(
            session,
            r"""import sys
rem = await remote.connect('fakehost')
fallback = await remote.connect('oldhost')
starts = [('local', run), ('remote', rem.run), ('fallback', fallback.run)]
expected = ('first file\n' + 'line αβ\n' * 5000 + 'second file\n' + 'other line\n' * 8000).encode()
program = "import sys; sys.stdout.buffer.write(('first file\\n' + 'line αβ\\n' * 5000 + 'second file\\n' + 'other line\\n' * 8000).encode())"
for label, start in starts:
    job = start(sys.executable, '-c', program)
    await job
    assert job.tail(60000) == expected.decode()[-60000:], label
    assert await job.head(40000) == expected.decode()[:40000], label
    assert await job.read() == expected.decode(), label
    assert job.tail(65536) == expected.decode()[-65536:], label
    assert await job.read(offset=17, limit=4000) == expected[17:4017].decode(errors='ignore'), label
    # Byte offsets remain byte offsets, even when a page cuts through UTF-8.
    offset = expected.index('α'.encode()) + 1
    assert await job.read(offset=offset, limit=9) == expected[offset:offset+9].decode(errors='ignore'), label
    for preview in (job.head, job.tail):
        try:
            await preview(3_000_000)
        except Exception as error:
            assert 'job.read()' in str(error) and 'job.save' in str(error), str(error)
        else:
            raise AssertionError('silently shortened a preview: ' + label)
    path = await job.save(label + '.patch')
    actual = await rem.read(path) if label == 'remote' else Path(path).read_bytes()
    if isinstance(actual, str):
        actual = actual.encode()
    assert actual == expected, label

async def refuses(call, *args, message, **kwargs):
    try:
        await call(*args, **kwargs)
    except Exception as error:
        assert message in str(error), str(error)
    else:
        raise AssertionError('accepted incomplete output')

for label, start in starts:
    job = start(sys.executable, '-c', "import sys; sys.stdout.buffer.write(b'begin\\xff\\x00' + b'x' * 140000 + b'end')")
    await job
    assert '\ufffd' in await job.read(), label
    assert await job.tail(0) == '' and await job.head(lines=0) == '', label
    await refuses(job.read, limit=65537, message='65536')
    # No newline in the window: a line request must not return a fragment.
    await refuses(job.tail, lines=1, message='job.read()')
    await refuses(job.head, lines=1, message='job.read()')
    path = await job.save(label + '.bin')
    actual = await rem.read(path) if label == 'remote' else Path(path).read_bytes()
    assert actual == b'begin\xff\x00' + b'x' * 140000 + b'end', label
    overflow = start(sys.executable, '-c', "import sys; sys.stdout.buffer.write(b'z' * (2097152 + 10) + b'\\xff\\x00end')")
    await overflow
    await refuses(overflow.read, message='paginate with')
    assert await overflow.read(offset=1048576, limit=10) == 'z' * 10, label
    assert await overflow.read(offset=1048576-2, limit=10) == 'z' * 10, label
    page_reader = rem.output.read if label == 'remote' else output.read
    assert await page_reader(overflow.id, offset=1048576, limit=10) == 'z' * 10, label
    pages = [await overflow.read(offset=offset, limit=65536)
             for offset in range(0, 2097152 + 15, 65536)]
    assert ''.join(pages) == 'z' * (2097152 + 10) + '\x00end', label
    assert await overflow.read(offset=2097152+15, limit=65536) == '', label
    await overflow.save(path)
    saved = await rem.read(path) if label == 'remote' else Path(path).read_bytes()
    assert saved == b'z' * (2097152 + 10) + b'\xff\x00end', label
    # The disk cap is explicit; stored pages remain accessible, lost pages refuse.
    capped = start(sys.executable, '-c', "import sys; sys.stdout.buffer.write(b'c' * (16777216 + 1))")
    await capped
    assert await capped.read(offset=16777217, limit=0) == '', label
    await refuses(capped.read, message='refusing incomplete output')
    await refuses(capped.save, path, message='refusing incomplete output')
    preserved = await rem.read(path) if label == 'remote' else Path(path).read_bytes()
    assert preserved == saved, 'overwrote destination with partial output: ' + label
    assert await capped.read(offset=16777216-10, limit=10) == 'c' * 10, label
    await refuses(capped.read, offset=16777216, limit=1, message='cannot be recovered by pagination')
    await refuses(page_reader, capped.id, offset=16777216-2, limit=10, message='cannot be recovered by pagination')
    unicode = start(sys.executable, '-c', "print('é' * 70000, end='')")
    await unicode
    await refuses(unicode.tail, 60000, message='job.read()')
    await refuses(unicode.head, 60000, message='job.read()')
    assert await unicode.tail(100) == 'é' * 100, label

# A running job must not overwrite a completed artifact.
protected = Path('protected.bin')
protected.write_bytes(b'keep me')
running = run(sys.executable, '-c', 'import time; time.sleep(0.5)')
await refuses(running.read, message='await job')
await refuses(running.save, protected, message='await job')
assert protected.read_bytes() == b'keep me'
await running
# Streaming to a file recovers output larger than the retained buffers.
streamed = run(sys.executable, '-c', "print('s' * 2000000, end='')").pipe('tee', 'whole.txt')
await streamed
assert Path('whole.txt').read_bytes() == b's' * 2000000
# A failed atomic rename leaves the destination and no temporary output file.
directory = Path('not-a-file')
directory.mkdir()
(directory / 'keep').write_bytes(b'preserved')
before = set(Path('.').iterdir())
await refuses(running.save, directory, message='directory')
assert set(Path('.').iterdir()) == before
assert (directory / 'keep').read_bytes() == b'preserved'
print('refusals preserve destinations and exact binary output')

await rem.close()
await fallback.close()
print('complete output preserved for all three job kinds')
""",
        )
        self.assertEqual(result["status"], "ok", result)
        self.assertIn("complete output preserved", result["output"])
        self.assertIn("refusals preserve destinations", result["output"])

    def test_job_spill_is_live_and_failures_preserve_artifacts(self):
        session = self.create(str(self.app.workspace))["id"]
        result = self.cell(
            session,
            r"""import albedo_capture
import tempfile
import time
from unittest.mock import patch

async def refuses(call, *args, message, **kwargs):
    try:
        await call(*args, **kwargs)
    except Exception as error:
        assert message in str(error), str(error)
    else:
        raise AssertionError('accepted incomplete output')

protected = Path('protected-spill.bin')
protected.write_bytes(b'keep me')
release = Path('release-writer').absolute()
program = f'''import sys, time
from pathlib import Path
sys.stdout.buffer.write(b'x' * 2097152)
sys.stdout.flush()
while not Path({str(release)!r}).exists():
    time.sleep(0.01)
sys.stdout.buffer.write(b'end')
sys.stdout.flush()
'''
live = run(sys.executable, '-c', program)
async def ready():
    while live.capture.seen < 2097152:
        await asyncio.sleep(0.01)
await asyncio.wait_for(ready(), 5)
assert live.duration is None
assert live.read(offset=1048576-2, limit=10) == 'x' * 10
assert output.read(live.id, offset=2097152-10, limit=10) == 'x' * 10
await refuses(live.save, protected, message='await job')
assert protected.read_bytes() == b'keep me'
assert Path(live.capture.spill).stat().st_mode & 0o077 == 0

# Simulate a disk write failure after a valid spill prefix has been captured.
with patch.object(live.capture._spill_file, 'write', side_effect=OSError('disk full')):
    release.touch()
    await live
assert live.capture._spill_file is None
assert live.tail(3) == 'end'
await refuses(live.read, offset=1048576, limit=1, message='refusing incomplete output')
await refuses(live.save, protected, message='refusing incomplete output')
assert protected.read_bytes() == b'keep me'

# Creation failure keeps previews usable, and never retries from a lost prefix.
blocked = Path('not-a-spill-directory')
blocked.write_text('blocked')
with patch.object(albedo_capture, 'SPILL_DIR', str(blocked)):
    failed = run(sys.executable, '-c', "print('f' * 2097152, end='')")
    await failed
assert failed.tail(10) == 'f' * 10
assert failed.capture.spill is None
await refuses(failed.read, offset=1048576, limit=1, message='refusing incomplete output')
await refuses(failed.save, protected, message='refusing incomplete output')
assert protected.read_bytes() == b'keep me'

# A missing or shortened spill must not become a short page or a partial save.
for damage in ('shorten', 'remove'):
    job = run(sys.executable, '-c', "print('s' * 2097152, end='')")
    await job
    assert job.capture._spill_file is None
    spill = Path(job.capture.spill)
    if damage == 'shorten':
        spill.write_bytes(b'short')
    else:
        spill.unlink()
    before = set(Path('.').iterdir())
    await refuses(job.read, offset=1048576, limit=10,
                  message='incomplete' if damage == 'shorten' else 'No such file')
    await refuses(job.save, protected,
                  message='incomplete' if damage == 'shorten' else 'No such file')
    assert protected.read_bytes() == b'keep me'
    assert set(Path('.').iterdir()) == before

# Pruning deletes only expired spill files, not fresh files or directories.
with tempfile.TemporaryDirectory() as directory:
    old = Path(directory, 'old.txt')
    old.write_text('old')
    fresh = Path(directory, 'fresh.txt')
    fresh.write_text('fresh')
    nested = Path(directory, 'nested')
    nested.mkdir()
    (nested / 'keep.txt').write_text('keep')
    stale = time.time() - albedo_capture.SPILL_KEEP - 1
    os.utime(old, (stale, stale))
    os.utime(nested, (stale, stale))
    previous = albedo_capture.SPILL_DIR
    try:
        albedo_capture.prune_spills(directory)
    finally:
        albedo_capture.SPILL_DIR = previous
    assert not old.exists() and fresh.read_text() == 'fresh'
    assert (nested / 'keep.txt').read_text() == 'keep'
print('live spill pages, disk failures, atomic saves, and pruning verified')
""",
        )
        self.assertEqual(result["status"], "ok", result)
        self.assertIn("pruning verified", result["output"])

    def test_a_first_visit_says_it_is_copying_the_kernel(self):
        self.slow_stage.touch()  # each copy of the bundle takes 3 s
        session = self.create(f"fakehost:{self.project}")["id"]
        seen = set()

        def watch():
            deadline = time.monotonic() + 15
            while time.monotonic() < deadline:
                with self.app.api(f"/sessions/{session}?tail=0") as response:
                    kernel = json.load(response)["kernel"]
                seen.add(kernel["stage"])
                if kernel["state"] == "attached":
                    return
                time.sleep(0.2)

        result = self.cell(session, "1", during=watch)
        self.assertEqual(result["status"], "ok", result)
        self.assertIn("staging", seen)

    def test_a_bridge_killed_mid_cell_reattaches_over_ssh(self):
        session = self.create(f"fakehost:{self.project}")["id"]
        first = self.cell(session, "import os\nos.getpid()")
        (bridge,) = processes(f"{self.remote_home}/.albedo-remote/", "albedo_bridge.py")

        def kill_bridge():
            time.sleep(1)
            os.kill(bridge, signal.SIGKILL)

        slept = self.cell(
            session,
            "import threading\nthreading.Event().wait(3)\nprint('woke')\nos.getpid()",
            during=kill_bridge,
        )
        self.assertEqual(slept["status"], "ok", slept)
        self.assertEqual(slept["output"].strip(), "woke")
        self.assertEqual(slept["value"], first["value"])
        bridges = [
            line
            for line in self.ssh_log.read_text().splitlines()
            if "albedo_bridge.py" in line
        ]
        self.assertGreaterEqual(len(bridges), 2, "the reattach did not go over ssh")

    def test_a_remote_kernel_outlives_a_daemon_restart(self):
        session = self.create(f"fakehost:{self.project}")["id"]
        first = self.cell(session, "import os\nsurvivor = 41\nos.getpid()")
        self.app.restart(crash=True)
        after = self.cell(session, "survivor += 1\n(survivor, os.getpid())")
        self.assertEqual(after["value"], f"(42, {first['value']})")

    def test_a_kernel_survives_ssh_being_down_while_the_daemon_restarts(self):
        session = self.create(f"fakehost:{self.project}")["id"]
        first = self.cell(session, "import os\nsurvivor = 41\nos.getpid()")
        self.down.touch()
        # Back a few seconds after the daemon came up, while its startup
        # attach and the turn's are still failing.
        threading.Timer(4, self.down.unlink).start()
        self.app.restart(crash=True)
        after = self.cell(session, "survivor += 1\n(survivor, os.getpid())")
        self.assertEqual(after["value"], f"(42, {first['value']})")

    def test_a_remote_sessions_memory_is_the_daemons_and_links_with_a_local_one(self):
        remote = f"fakehost:{self.project}"
        session = self.create(remote)["id"]
        self.cell(session, "memory.append('the gate runs on fakehost')")
        slug = re.sub(r"[^A-Za-z0-9]", "-", remote)
        written = self.app.home / "memories" / slug / "memory.md"
        self.assertEqual(written.read_text(), "the gate runs on fakehost\n")
        self.assertFalse((self.remote_home / ".albedo").exists())

        # The same project checked out here, linked: it reads the remote one.
        checkout = self.app.root / "checkout"
        checkout.mkdir()
        local = self.create(str(checkout))["id"]
        merged = self.merge(checkout, remote)
        self.assertIn(remote, merged["resource"]["value"]["members"])
        found = self.cell(local, "print(memory.grep('gate'))")
        self.assertIn(f"[{remote}] memory.md:1: the gate runs", found["output"])

    def test_a_linked_remote_folder_is_checked_on_its_host(self):
        checkout = self.app.root / "checkout"
        checkout.mkdir()
        self.create(str(checkout))

        def presence():
            return {
                item["workspace"]: item for item in self.group(checkout)["presence"]
            }

        present = f"fakehost:{self.project}"
        missing = f"fakehost:{self.remote_home}/missing"
        for member in (present, missing, "gone:/srv/app"):
            self.merge(checkout, member)
        found = presence()
        self.assertTrue(found[present]["exists"])
        self.assertFalse(found[missing]["exists"])
        self.assertEqual(found["gone:/srv/app"]["host"]["state"], "unreachable")
        self.assertIn(
            "Connection refused", found["gone:/srv/app"]["host"]["detail"]["detail"]
        )

        # A host that stops answering is not a gone folder.
        self.down.touch()
        found = presence()
        self.assertIsNone(found[present]["exists"])
        self.assertIsNone(found[missing]["exists"])
        with self.app.api("/hosts/fakehost/probe", {}) as response:
            self.assertEqual(response.status, 202)
        deadline = time.monotonic() + 15
        while (
            self.host("fakehost")["state"] == "probing" and time.monotonic() < deadline
        ):
            time.sleep(0.05)
        found = presence()
        self.assertEqual(found[present]["host"]["state"], "unreachable")
        self.assertEqual(found[missing]["host"]["state"], "unreachable")
        self.down.unlink()
        with self.app.api("/hosts/fakehost/probe", {}) as response:
            self.assertEqual(response.status, 202)
        deadline = time.monotonic() + 15
        while (
            self.host("fakehost")["state"] == "probing" and time.monotonic() < deadline
        ):
            time.sleep(0.05)
        self.assertEqual(self.host("fakehost")["state"], "ready")

        # A session opening with the host's probe fresh is told what is gone.
        self.cell(self.create(str(checkout))["id"], "1")
        instructions = self.provider.requests[-1]["request"]["instructions"]
        self.assertIn(f"{missing} (its folder is gone)", instructions)
        self.assertNotIn(f"{present} (its folder is gone)", instructions)

    def test_linked_hosts_are_checked_side_by_side(self):
        checkout = self.app.root / "slow-checkout"
        checkout.mkdir()
        self.create(str(checkout))

        for host in ("slow-a", "slow-b"):
            self.merge(checkout, f"{host}:/srv")
        # Each host takes 3 s to refuse; one after the other would be 6.
        began = time.monotonic()
        rows = [
            item
            for item in self.group(checkout)["presence"]
            if item["workspace"] != str(checkout)
        ]
        self.assertLess(time.monotonic() - began, 5.5)
        self.assertEqual([row["host"]["state"] for row in rows], ["unreachable"] * 2)

    def test_a_remote_home_is_resolved_and_stored_absolute(self):
        created = self.create("fakehost:~/proj")
        self.assertEqual(created["workspace"], f"fakehost:{self.project}")
        with self.assertRaises(urllib.error.HTTPError) as refused:
            self.create("gone:~/proj")
        self.assertEqual(refused.exception.code, 503)
        self.assertIn("Connection refused", refused.exception.read().decode())

    def test_hosts_report_what_ssh_found(self):
        def warm(host):
            with self.app.api(f"/hosts/{host}/probe", {}) as response:
                self.assertEqual(response.status, 202)
                observation = json.load(response)
            deadline = time.monotonic() + 15
            while observation["state"] == "probing" and time.monotonic() < deadline:
                time.sleep(0.05)
                observation = self.host(host)
            self.assertNotEqual(observation["state"], "probing", observation)
            return observation

        ready = warm("fakehost")
        self.assertEqual(ready["state"], "ready", ready)
        self.assertEqual(ready["home"], str(self.remote_home))
        self.assertTrue(ready["os"] and ready["architecture"])
        old = warm("oldhost")
        self.assertEqual(old["state"], "unsupported", old)
        self.assertIn("python >= 3.11", old["detail"]["detail"])
        self.assertEqual(warm("locked")["state"], "needs_auth")
        self.assertEqual(warm("gone")["state"], "unreachable")
        # A host whose ssh config names a ControlPath rides that master, so the
        # user's own ssh and albedo share one sign-in.
        self.assertEqual(warm("muxed")["state"], "ready")
        muxed = [
            line
            for line in self.ssh_log.read_text().splitlines()
            if " muxed " in line and "-G" not in line
        ]
        self.assertTrue(muxed)
        for line in muxed:
            self.assertIn(f"ControlPath={self.mux} ", line)
            self.assertIn("ControlPersist=1800 ", line)
        self.assertEqual(self.host("fakehost")["state"], "ready")

        # A bracketed IPv6 host travels percent-encoded in the probe's path and
        # is probed as itself, the same host a target query names.
        with self.app.api("/hosts/%5B::1%5D/probe", {}) as response:
            self.assertEqual(response.status, 202)
            self.assertEqual(json.load(response)["target"], "[::1]")
        self.host("[::1]")

        # A turn at a host that needs a sign-in waits with ssh's words, and
        # the host names the control path the tui's sign-in opens.
        session = self.create("locked:/srv/app")["id"]
        with self.app.prompt(session, "hello") as response:
            input_id = json.load(response)["id"]
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            with self.app.api(f"/sessions/{session}/inputs/{input_id}") as response:
                receipt = json.load(response)
            if receipt["blocking_reason"]:
                break
            time.sleep(0.05)
        self.assertEqual(receipt["delivery"], "pending", receipt)
        self.assertIn("Permission denied", receipt["blocking_reason"]["detail"])
        host = self.host("locked")
        self.assertEqual(host["state"], "needs_auth", host)
        self.assertTrue(host["authentication"]["control_path"].endswith("/%C"), host)

    def test_a_remote_kernel_is_ended_on_its_host_never_here(self):
        session = self.create(f"fakehost:{self.project}")["id"]
        started = self.cell(
            session,
            "import asyncio\njob = run('sleep', '7777')\nawait asyncio.sleep(0.5)\njob.exit_code",
        )
        self.assertEqual(started["status"], "ok", started)
        self.assertTrue(processes("sleep 7777"))
        elsewhere = self.remote_home / "elsewhere"
        elsewhere.mkdir()
        with self.app.api(f"/sessions/{session}?view=configuration") as response:
            configuration = json.load(response)
            revision = response.headers["ETag"]
        self.app.api(
            f"/sessions/{session}?view=configuration",
            {
                "workspace": f"fakehost:{elsewhere}",
                "family_revision": configuration["family_revision"],
            },
            method="PATCH",
            headers={"If-Match": revision},
        ).close()
        deadline = time.monotonic() + 15
        while processes("sleep 7777") and time.monotonic() < deadline:
            time.sleep(0.1)
        self.assertFalse(processes("sleep 7777"))
        remote_ladder = [
            line
            for line in self.ssh_log.read_text().splitlines()
            if "albedo_signal.py" in line
        ]
        self.assertTrue(remote_ladder, "the kernel was not ended over ssh")
        local = self.python_log.read_text() if self.python_log.exists() else ""
        self.assertNotIn("albedo_signal.py", local)

    def test_jobs_started_over_a_remote_connection_are_listed_like_local_ones(self):
        session = self.create(str(self.app.workspace))["id"]
        started = self.cell(
            session,
            "import asyncio\n"
            "rem = await remote.connect('fakehost')\n"
            "fallback = await remote.connect('oldhost')\n"
            "rem.run('sleep', '8881')\n"
            "fallback.run('sleep', '6')\n"
            "await asyncio.sleep(1)",
        )
        self.assertEqual(started["status"], "ok", started)
        kernel = json.load(self.app.api(f"/sessions/{session}?tail=0"))["kernel"]
        self.assertEqual(kernel["live_job_count"], 2, kernel)
        running = {job["command"]: job for job in kernel["running_jobs"]}
        self.assertEqual(sorted(running), ["sleep 6", "sleep 8881"], kernel)
        for job in running.values():
            self.assertLess(abs(job["started_at"] - time.time() * 1000), 60_000)
        with self.app.api(f"/extensions/run/sessions/{session}/jobs") as response:
            page = json.load(response)["page"]
        self.assertEqual(page["summary"], "2 background jobs running")
        self.assertEqual(
            sorted(row["text"] for row in page["rows"]), ["sleep 6", "sleep 8881"]
        )
        ended = self.cell(session, "await rem.close()\nawait asyncio.sleep(1)")
        self.assertEqual(ended["status"], "ok", ended)
        kernel = json.load(self.app.api(f"/sessions/{session}?tail=0"))["kernel"]
        self.assertNotIn(
            "sleep 8881", [job["command"] for job in kernel["running_jobs"]]
        )

    def test_remote_background_jobs_can_be_stopped_through_run_api(self):
        session = self.create(str(self.app.workspace))["id"]
        started = self.cell(
            session,
            "import asyncio\n"
            "rem = await remote.connect('fakehost')\n"
            "fallback = await remote.connect('oldhost')\n"
            "rem_job = rem.run('sleep', '8892')\n"
            "fallback_job = fallback.run('sleep', '8893')\n"
            "await asyncio.sleep(1)",
        )
        self.assertEqual(started["status"], "ok", started)

        def cleanup():
            try:
                self.cell(
                    session,
                    "try:\n"
                    "    await rem_job.stop()\n"
                    "except Exception:\n"
                    "    pass\n"
                    "try:\n"
                    "    await fallback_job.stop()\n"
                    "except Exception:\n"
                    "    pass\n"
                    "try:\n"
                    "    await rem.close()\n"
                    "except Exception:\n"
                    "    pass\n"
                    "try:\n"
                    "    await fallback.close()\n"
                    "except Exception:\n"
                    "    pass\n",
                )
            except Exception:
                pass
            for pid in processes("sleep 8892") + processes("sleep 8893"):
                try:
                    os.kill(pid, signal.SIGKILL)
                except OSError:
                    pass

        self.addCleanup(cleanup)

        kernel = json.load(self.app.api(f"/sessions/{session}?tail=0"))["kernel"]
        self.assertEqual(kernel["live_job_count"], 2, kernel)
        running = {job["command"]: job for job in kernel["running_jobs"]}
        self.assertEqual(sorted(running), ["sleep 8892", "sleep 8893"], kernel)
        rem_job_id = running["sleep 8892"]["id"]
        fallback_job_id = running["sleep 8893"]["id"]
        self.assertTrue(wait_for(lambda: processes("sleep 8892"), timeout=5))

        with self.app.api(f"/extensions/run/sessions/{session}/jobs") as response:
            page = json.load(response)["page"]
        self.assertEqual(page["summary"], "2 background jobs running")
        self.assertEqual(
            sorted(row["text"] for row in page["rows"]), ["sleep 8892", "sleep 8893"]
        )

        # 3. Stopping an unknown job id still fails with 404
        with self.assertRaises(urllib.error.HTTPError) as missing:
            self.app.api(
                f"/extensions/run/sessions/{session}/jobs/missing/stop",
                method="POST",
            )
        self.assertEqual(missing.exception.code, 404)

        # 1. Stop the kernel-mode remote job through the run service API
        with self.app.api(
            f"/extensions/run/sessions/{session}/jobs/{rem_job_id}/stop",
            method="POST",
        ) as response:
            self.assertEqual(response.status, 200)
            self.assertEqual(json.load(response), {"stopped": rem_job_id})

        wait_for(
            lambda: not processes("sleep 8892"),
            timeout=15,
            message="remote sleep 8892 process did not terminate after stop",
        )

        def rem_cleared_from_kernel():
            with self.app.api(f"/sessions/{session}?tail=0") as resp:
                k = json.load(resp)["kernel"]
            ids = [job["id"] for job in k["running_jobs"]]
            return rem_job_id not in ids and k["live_job_count"] == 1

        wait_for(
            rem_cleared_from_kernel,
            timeout=15,
            message="kernel observation did not drop stopped remote job",
        )

        def rem_cleared_from_page():
            with self.app.api(f"/extensions/run/sessions/{session}/jobs") as resp:
                p = json.load(resp)["page"]
            ids = [row["id"] for row in p["rows"]]
            return rem_job_id not in ids and p["summary"] == "1 background job running"

        wait_for(
            rem_cleared_from_page,
            timeout=15,
            message="jobs page did not drop stopped remote job",
        )

        # 2. Stop the degraded fallback job through the run service API
        with self.app.api(
            f"/extensions/run/sessions/{session}/jobs/{fallback_job_id}/stop",
            method="POST",
        ) as response:
            self.assertEqual(response.status, 200)
            self.assertEqual(json.load(response), {"stopped": fallback_job_id})

        def fallback_cleared_from_kernel():
            with self.app.api(f"/sessions/{session}?tail=0") as resp:
                k = json.load(resp)["kernel"]
            ids = [job["id"] for job in k["running_jobs"]]
            return fallback_job_id not in ids and k["live_job_count"] == 0

        wait_for(
            fallback_cleared_from_kernel,
            timeout=15,
            message="kernel observation did not drop stopped fallback job",
        )

        def fallback_cleared_from_page():
            with self.app.api(f"/extensions/run/sessions/{session}/jobs") as resp:
                p = json.load(resp)["page"]
            ids = [row["id"] for row in p["rows"]]
            return fallback_job_id not in ids and p["rows"] == []

        wait_for(
            fallback_cleared_from_page,
            timeout=15,
            message="jobs page did not drop stopped fallback job",
        )

        output = self.cell(session, "fallback_job.tail()")
        self.assertEqual(output["status"], "ok", output)
        self.assertIn(
            "the remote command may still be running",
            output.get("value", "") + output.get("output", ""),
        )

        ended = self.cell(
            session,
            "await rem.close()\nawait fallback.close()\nawait asyncio.sleep(0.5)",
        )
        self.assertEqual(ended["status"], "ok", ended)

    def test_hosts_list_recent_and_configured_hosts_without_probing(self):
        self.create(f"fakehost:{self.project}")
        with self.app.api("/hosts") as response:
            hosts = json.load(response)["items"]
        by_host = {entry["target"]: entry for entry in hosts}
        self.assertTrue({"fakehost", "configured-box", "included-box"} <= set(by_host))
        self.assertNotIn("*", by_host)
        self.assertFalse([host for host in by_host if "*" in host])
        # Nothing was probed just to list them.
        self.assertEqual(by_host["configured-box"]["state"], "unknown")
        self.assertIsNone(by_host["configured-box"]["observed_at"])

    def browse(self, location, *, preview=False):
        query = {"location": str(location)}
        if preview:
            query["include"] = "preview"
        with self.app.api("/workspaces?" + urllib.parse.urlencode(query)) as response:
            return json.load(response)

    def assert_same_answers(self, tree):
        """Remote and local inspection preserve the same filesystem facts."""
        remote = self.browse(f"fakehost:{tree}", preview=True)
        local = self.browse(tree, preview=True)
        self.assertEqual(remote["directory"], f"fakehost:{tree}")
        self.assertEqual(remote["home"], f"fakehost:{self.remote_home}")
        self.assertEqual(
            [
                {key: row[key] for key in ("name", "vcs", "hidden", "modified_at")}
                for row in remote["items"]
            ],
            [
                {key: row[key] for key in ("name", "vcs", "hidden", "modified_at")}
                for row in local["items"]
            ],
        )
        remote_preview = remote["preview"]
        local_preview = local["preview"]
        for key in ("languages", "more"):
            self.assertEqual(remote_preview[key], local_preview[key])
        self.assertEqual(len(remote_preview["tree"]), len(local_preview["tree"]))
        for remote_row, local_row in zip(remote_preview["tree"], local_preview["tree"]):
            self.assertEqual(
                remote_row["location"], "fakehost:" + local_row["location"]
            )
            self.assertEqual(
                {
                    key: value
                    for key, value in remote_row.items()
                    if key not in {"location", "children"}
                },
                {
                    key: value
                    for key, value in local_row.items()
                    if key not in {"location", "children"}
                },
            )
            self.assertEqual(len(remote_row["children"]), len(local_row["children"]))
            for remote_child, local_child in zip(
                remote_row["children"], local_row["children"]
            ):
                self.assertEqual(
                    remote_child["location"], "fakehost:" + local_child["location"]
                )
                self.assertEqual(
                    {
                        key: value
                        for key, value in remote_child.items()
                        if key != "location"
                    },
                    {
                        key: value
                        for key, value in local_child.items()
                        if key != "location"
                    },
                )
        remote_repo = remote_preview["repository"]
        local_repo = local_preview["repository"]
        self.assertEqual(remote_repo is None, local_repo is None)
        if remote_repo is not None:
            self.assertEqual(remote_repo["root"], "fakehost:" + local_repo["root"])
            self.assertEqual(
                {key: value for key, value in remote_repo.items() if key != "root"},
                {key: value for key, value in local_repo.items() if key != "root"},
            )

    def make_tree(self, name):
        tree = self.remote_home / name
        (tree / "src" / "deep").mkdir(parents=True)
        (tree / "docs").mkdir()
        (tree / "src" / "main.py").write_text("print('hi')\n" * 40)
        (tree / "src" / "deep" / "util.go").write_text("package deep\n")
        (tree / "README.md").write_text("# tree\n")
        (tree / ".hidden").mkdir()
        return tree

    def test_a_remote_git_repo_browses_like_a_local_one(self):
        if not shutil.which("git"):
            self.skipTest("git is not installed")
        tree = self.make_tree("git-tree")
        git = ["git", "-c", "user.name=t", "-c", "user.email=t@e", "-C", str(tree)]
        subprocess.run([*git, "init", "-q", "-b", "main"], check=True)
        subprocess.run([*git, "add", "src", "README.md"], check=True)
        subprocess.run([*git, "commit", "-qm", "one"], check=True)
        (tree / "docs" / "new.md").write_text("new\n")
        self.assertEqual(
            self.browse(f"fakehost:{tree}", preview=True)["preview"]["repository"][
                "kind"
            ],
            "git",
        )
        self.assert_same_answers(tree)
        self.assert_same_answers(tree / "src")

    def test_a_remote_jj_repo_browses_like_a_local_one(self):
        if not shutil.which("jj"):
            self.skipTest("jj is not installed")
        tree = self.make_tree("jj-tree")
        env = {**os.environ, "JJ_USER": "t", "JJ_EMAIL": "t@e"}
        subprocess.run(["jj", "git", "init", "--quiet", str(tree)], check=True, env=env)
        subprocess.run(
            ["jj", "-R", str(tree), "bookmark", "create", "main", "-r", "@", "--quiet"],
            check=True,
            env=env,
        )
        self.assertEqual(
            self.browse(f"fakehost:{tree}", preview=True)["preview"]["repository"][
                "kind"
            ],
            "jj",
        )
        self.assert_same_answers(tree)

    def test_a_remote_folder_answers_with_the_hosts_state(self):
        query = urllib.parse.urlencode({"location": "gone:/srv"})
        with self.assertRaises(urllib.error.HTTPError) as refused:
            self.app.api(f"/workspaces?{query}").close()
        self.assertEqual(refused.exception.code, 503)
        body = json.loads(refused.exception.read())
        self.assertEqual(body["status"], 503)
        self.assertIn("Connection refused", body["detail"])
        self.assertEqual(self.host("gone")["state"], "unreachable")
        # A session there still composes, its project files skipped.
        session = self.create("gone:/srv/app")["id"]
        with self.app.api(f"/sessions/{session}/catalog") as response:
            self.assertTrue(json.load(response)["discovery"]["candidates"])
        # A ~ path lists the remote home.
        listed = self.browse("fakehost:~")
        self.assertEqual(listed["directory"], f"fakehost:{self.remote_home}")
        self.assertIn("proj", [entry["name"] for entry in listed["items"]])

    def test_project_instructions_and_skills_come_from_the_host(self):
        (self.project / "AGENTS.md").write_text("remote rule: always say banana\n")
        skill = self.project / ".agents" / "skills" / "far-skill"
        skill.mkdir(parents=True)
        (skill / "SKILL.md").write_text(
            "---\nname: far-skill\ndescription: a skill that lives on the host\n---\nbody\n"
        )
        session = self.create(f"fakehost:{self.project}")["id"]
        self.app.prompt(session, "hello").close()
        self.app.idle(session, timeout=60)
        request = json.dumps(self.provider.requests[-1]["request"])
        self.assertIn("always say banana", request)
        self.assertIn("a skill that lives on the host", request)
        self.assertNotIn("aren't available yet", request)


if __name__ == "__main__":
    unittest.main()