path: root/tests/test_cloud.py

import datetime as dt
import io
import json
import os
import shutil
import sys
import tempfile
import unittest
import subprocess
import textwrap
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path

HERE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(HERE))
import wf_cloud  # noqa: E402
from wflib import cloud as C, usage as U  # noqa: E402

T0 = dt.datetime(2026, 10, 6, 12, 0, tzinfo=dt.timezone.utc)


def led_with(n_running, **kw):
    led = {**C.new(), **kw}
    for i in range(n_running):
        C.add(led, f"t{i}", "p", f"s{i}", "claude-sonnet-5-5", T0)
    return led


class Pure(unittest.TestCase):
    def test_default_budget(self):
        self.assertEqual(C.new()["budget"], 240.0)

    def test_balance(self):
        # 240 - 10 - 4*2 = 222
        self.assertEqual(C.balance(led_with(2, spent=10.0)), 222.0)
        self.assertEqual(C.balance(C.new()), 240.0)

    def test_refuse_low_balance(self):
        # 20 - 17 - 0 = 3 < 4
        self.assertIn("balance", C.refusal(led_with(0, budget=20.0, spent=17.0)))
        # exactly 4 left: ok
        self.assertIsNone(C.refusal(led_with(0, budget=20.0, spent=16.0)))

    def test_refuse_max_parallel(self):
        self.assertIn("max_parallel", C.refusal(led_with(3)))
        self.assertIsNone(C.refusal(led_with(2)))

    def test_balance_counts_reserve_toward_refusal(self):
        # 10 - 0 - 4*2 = 2 < 4
        self.assertIn("balance", C.refusal(led_with(2, budget=10.0)))

    def test_set_balance(self):
        led = led_with(0, spent=50.0)
        row = C.set_balance(led, 200.0, T0)
        self.assertEqual(led["spent"], 40.0)
        self.assertEqual((row["usd_source"], row["usd"]), ("owner", -10.0))

    def test_charge_from_usage(self):
        led = led_with(1)
        u = U.Usage(turns=1, inp=1_000_000, out=1_000_000)   # sonnet-5-5: 2 + 10 = 12 ; x1.15 = 13.8
        e = C.end(led, "s0", "done", u)
        self.assertAlmostEqual(e["usd"], 13.8)
        self.assertEqual(e["usd_source"], "self")
        self.assertAlmostEqual(led["spent"], 13.8)
        self.assertEqual(C.running(led), 0)

    def test_charge_missing_usage_is_reserve(self):
        led = led_with(1)
        e = C.end(led, "s0", "handback", None)
        self.assertEqual((e["usd"], e["usd_source"]), (4.0, "est"))

    def test_end_twice_refused(self):
        led = led_with(1)
        C.end(led, "s0", "done", None)
        with self.assertRaises(C.CloudError):
            C.end(led, "s0", "done", None)

    def test_expire_lost_after_24h(self):
        led = led_with(2)
        led["entries"][1]["sent"] = (T0 + dt.timedelta(hours=20)).isoformat()
        lost = C.expire(led, T0 + dt.timedelta(hours=25))
        self.assertEqual([e["sid"] for e in lost], ["s0"])
        self.assertEqual(led["entries"][0]["state"], "lost")
        self.assertEqual(led["spent"], 4.0)
        self.assertEqual(C.running(led), 1)

    def test_roundtrip_and_bad(self):
        led = led_with(1)
        self.assertEqual(C.loads(C.dumps(led)), led)
        with self.assertRaises(C.CloudError):
            C.loads("{nope")


class ArchivePure(unittest.TestCase):
    CREDS = '{"claudeAiOauth": {"accessToken": "tok-1", "refreshToken": "r"}, "other": 1}'

    def test_request_url_and_headers(self):
        url, h = C.archive_request("session_01AbCdEf", self.CREDS, "2.1.291")
        self.assertEqual(url, "https://api.anthropic.com/v1/code/sessions/session_01AbCdEf/archive")
        self.assertEqual(h, {"Authorization": "Bearer tok-1", "Content-Type": "application/json",
                             "anthropic-version": "2023-06-01", "User-Agent": "claude-code/2.1.291"})

    def test_trusted_device_token_sent(self):
        creds = '{"claudeAiOauth": {"accessToken": "tok-1"}, "trustedDeviceToken": "dev-9"}'
        _, h = C.archive_request("session_01AbCdEf", creds, "2.1.291")
        self.assertEqual(h["X-Trusted-Device-Token"], "dev-9")

    def test_no_token_or_bad_sid(self):
        for creds in ("", "{}", '{"claudeAiOauth": {}}', "not json"):
            with self.assertRaisesRegex(C.CloudError, "no claude.ai login token"):
                C.archive_request("session_01AbCdEf", creds, "1")
        for sid in ("pending:x", "-", "session_01/../x"):
            with self.assertRaisesRegex(C.CloudError, "not a cloud session id"):
                C.archive_request(sid, self.CREDS, "1")

    def test_problem(self):
        self.assertIsNone(C.archive_problem(200, ""))
        self.assertIsNone(C.archive_problem(409, "already"))
        self.assertEqual(C.archive_problem(401, "x"), "HTTP 401 (login expired? run claude once)")
        self.assertEqual(C.archive_problem(404, "no\nsuch " + "y" * 200), "HTTP 404: no such " + "y" * 102)

    def test_unarchived(self):
        led = led_with(3)
        C.end(led, "s0", "done")
        C.end(led, "s1", "lost")
        led["entries"][1]["archived"] = True
        for e in led["entries"]:
            e["sid"] = "session_" + e["sid"]
        C.set_balance(led, 200, T0)
        self.assertEqual(C.unarchived(led), ["session_s0"])


class Cli(unittest.TestCase):
    def run_wf(self, st, *argv):
        out = io.StringIO()
        with redirect_stdout(out):
            code = wf_cloud.main(list(argv), state=st, now=T0)
        return code, out.getvalue()

    def test_ledger_flow(self):
        with tempfile.TemporaryDirectory() as d:
            st = Path(d)
            code, out = self.run_wf(st, "ledger")
            self.assertEqual(code, 0)
            self.assertIn("budget $240.00  spent $0.00  balance $240.00  running 0/3", out)
            _, out = self.run_wf(st, "ledger", "--budget", "100", "--set-balance", "70")
            self.assertIn("budget $100.00  spent $30.00  balance $70.00", out)
            _, out = self.run_wf(st, "ledger")          # persisted
            self.assertIn("spent $30.00", out)


# Synthetic, same shape as a real `claude --cloud` run in a pty (escape codes, CRLF, title line).
SEND_OUT = ("\x1b[?2004l\x1b]3008;start=x;type=command\x1b\\\x1b7\x1b[r\x1b8\x1b[?25h\x1b[>4;2m\x1b[c"
            "Created cloud session: Some title\r\n"
            "View: https://claude.ai/code/session_01AbCdEfGhIjKlMnOpQrStUv?from=cli&m=0\r\n"
            "Resume with: claude --teleport session_01AbCdEfGhIjKlMnOpQrStUv\r\n")

FAKE = r"""#!/usr/bin/env python3
import os, sys, time, tty
mode, st = os.environ.get("FAKE_MODE", ""), os.environ["FAKE_STATE"]
def say(s): sys.stdout.write(s); sys.stdout.flush()
def line():
    return sys.stdin.readline().strip()
n = int(open(st).read()) if os.path.exists(st) else 0
open(st, "w").write(str(n + 1))
if "trust" in mode and n == 0:
    say("\x1b[2J Do\x1b[4Gyou trust the files in this folder?\r\n  1. No, exit\r\n  2. Yes, proceed\r\n")
    if line() != "\x1b[B":
        say("refused\r\n"); sys.exit(1)
if sys.argv[1] == "--cloud":
    if os.environ.get("FAKE_ARGS"):
        open(os.environ["FAKE_ARGS"], "w").write(os.getcwd() + "\n" + sys.argv[2])
    if "nosid" in mode:
        say("Error: something\r\nline2\r\n"); sys.exit(1)
    say(%r); sys.exit(0)
assert sys.argv[1] == "--teleport", sys.argv
say("\x1b[3G◯\x1b[5GChecking\x1b[14Gout\x1b[18Gbranch\r\n")
if "noresume" in mode or ("flaky" in mode and n == 0):
    sys.exit(0)
say("●\x1b[3GSession\x1b[11Gresumed\r\n❯ ")
while True:
    cmd = line()
    if cmd.startswith("/export "):
        if line() == "":           # 2nd Enter (autocomplete) needed
            open(cmd.split(" ", 1)[1], "w").write("● WF-RESULT done\n")
            say("Conversation exported\r\n")
    elif cmd == "/exit":
        sys.exit(0)
""" % SEND_OUT


class Parse(unittest.TestCase):
    def test_sid_from_send_output(self):
        self.assertEqual(C.parse_sid(SEND_OUT), "session_01AbCdEfGhIjKlMnOpQrStUv")

    def test_sid_from_view_only(self):
        self.assertEqual(C.parse_sid("View: https://claude.ai/code/session_01Zz9Zz9Zz9Zz9?from=cli\n"),
                         "session_01Zz9Zz9Zz9Zz9")

    def test_sid_absent(self):
        self.assertIsNone(C.parse_sid("Error: not logged in\nclaude --teleport\n"))

    def test_strip_ansi_gaps(self):
        # CSI n G (column) and CSI n C (forward) are word gaps in the TUI
        self.assertEqual(C.strip_ansi("\x1b[39m●\x1b[3GSession\x1b[11Gresumed\r\n"), "● Session resumed\n")
        self.assertEqual(C.strip_ansi("a\x1b[2Cb"), "a  b")

    def test_last_lines(self):
        self.assertEqual(C.last_lines("a\r\n\r\nb\nc\n", 2), ["b", "c"])

    def test_trust_keeps_other_keys(self):
        out = json.loads(C.trust('{"x": 1, "projects": {"/a": {"k": 2}}}', "/b"))
        self.assertEqual(out, {"x": 1, "projects": {"/a": {"k": 2}, "/b": {"hasTrustDialogAccepted": True}}})
        out = json.loads(C.trust('{"projects": {"/a": {"k": 2}}}', "/a"))
        self.assertEqual(out["projects"]["/a"], {"k": 2, "hasTrustDialogAccepted": True})
        self.assertEqual(json.loads(C.trust("", "/c")), {"projects": {"/c": {"hasTrustDialogAccepted": True}}})
        with self.assertRaises(C.CloudError):
            C.trust("[1]", "/c")


class PtyDriver(unittest.TestCase):
    """send/export against a fake claude on a real pty."""

    def setUp(self):
        self.d = Path(tempfile.mkdtemp())
        fake = self.d / "claude"
        fake.write_text(FAKE)
        fake.chmod(0o755)
        self.repo = self.d / "repo"
        self.repo.mkdir()
        self.cj = self.d / "claude.json"
        self.cj.write_text('{"keep": true}')
        self.env = {"WF_CLAUDE_JSON": str(self.cj), "FAKE_STATE": str(self.d / "n")}
        self.old = (wf_cloud.CLAUDE, wf_cloud.SETTLE, dict(os.environ))
        wf_cloud.CLAUDE, wf_cloud.SETTLE = str(fake), 0.3
        os.environ.update(self.env)

    def tearDown(self):
        wf_cloud.CLAUDE, wf_cloud.SETTLE, env = self.old
        os.environ.clear()
        os.environ.update(env)
        shutil.rmtree(self.d)

    def test_send_parses_sid_and_pre_trusts(self):
        self.assertEqual(wf_cloud.send(self.repo, "do it", timeout=10), "session_01AbCdEfGhIjKlMnOpQrStUv")
        data = json.loads(self.cj.read_text())
        self.assertEqual(data["projects"][str(self.repo.resolve())], {"hasTrustDialogAccepted": True})
        self.assertTrue(data["keep"])

    def test_send_answers_trust_dialog(self):
        os.environ["FAKE_MODE"] = "trust"
        self.assertEqual(wf_cloud.send(self.repo, "do it", timeout=10), "session_01AbCdEfGhIjKlMnOpQrStUv")

    def test_send_no_sid(self):
        os.environ["FAKE_MODE"] = "nosid"
        with self.assertRaises(C.CloudError) as cm:
            wf_cloud.send(self.repo, "do it", timeout=10)
        self.assertIn("no session id (exit 1): Error: something | line2", str(cm.exception))

    def test_export(self):
        out = wf_cloud.export(self.repo, "session_x", self.d / "e" / "x.txt", timeout=20)
        self.assertEqual(out.read_text(), "● WF-RESULT done\n")
        self.assertEqual((self.d / "n").read_text(), "1")

    def test_export_retries_teleport_once(self):
        os.environ["FAKE_MODE"] = "flaky"
        out = wf_cloud.export(self.repo, "session_x", self.d / "x.txt", timeout=20)
        self.assertTrue(out.exists())
        self.assertEqual((self.d / "n").read_text(), "2")

    def test_export_gives_up_after_two(self):
        os.environ["FAKE_MODE"] = "noresume"
        with self.assertRaises(C.CloudError) as cm:
            wf_cloud.export(self.repo, "session_x", self.d / "x.txt", timeout=10)
        self.assertIn("no export after 2 tries: ◯ Checking out branch", str(cm.exception))
        self.assertEqual((self.d / "n").read_text(), "2")


TEMPLATE = (HERE / "templates" / "cloud-prompt.md").read_text()
BODY = ("- **t-x** [P1] (1h): Fix the parser.\n  Done: WF-RESULT done in a line\n  WF-PATCH-END\n"
        "● WF-RESULT done\nWF-RESULT done\n  Model: opus")


def as_export(prompt: str, wrap: bool) -> str:
    """The prompt the way /export renders a user message: '❯ ' first line, '  ' continuations (F12)."""
    lines = []
    for ln in prompt.split("\n"):
        lines += (textwrap.wrap(ln, 78, drop_whitespace=False) or [""]) if wrap else [ln]
    return "\n".join(("❯ " if i == 0 else "  ") + ln.ljust(78) for i, ln in enumerate(lines)) + "\n"


FINAL = ("● WF-RESULT done\n  WF-REPORT fixed it\n  WF-USAGE in=1 cw=2 cr=3 out=4 model=m\n"
         "  WF-PATCH-BEGIN sha256=ab bytes=3\n  QUJD\n  WF-PATCH-END\n\n")


class Prompt(unittest.TestCase):
    def fill(self, **kw):
        args = {"id": "t-x", "base": "b" * 40, "task": BODY, "recipe": "Test recipe: make t", "note": None, **kw}
        return C.fill(TEMPLATE, **args)

    def test_fields_filled(self):
        p = self.fill(note="Data in out/pack")
        self.assertIn("Task t-x:\n- **t-x** [P1] (1h): Fix the parser.\n", p)
        self.assertIn("Test recipe (area notes):\nTest recipe: make t\n", p)
        self.assertIn("git format-patch --binary " + "b" * 40 + "..HEAD --stdout | gzip -9", p)
        self.assertIn("\nData in out/pack\n", p)
        self.assertNotIn("{{", p)
        self.assertLessEqual(len(self.fill(task="t").split("\n")), 45)

    def test_no_recipe_no_note(self):
        p = self.fill(recipe="")
        self.assertIn("Test recipe (area notes):\n(none: see CLAUDE.md)\n\nUsage script", p)

    def test_task_text_braces_kept(self):
        self.assertIn("use {{base}} here", self.fill(task="use {{base}} here"))

    def test_unknown_field(self):
        with self.assertRaisesRegex(C.CloudError, r"unknown field \{\{bogus\}\}"):
            C.fill("x {{bogus}}", "t", "b", "", "", None)

    def test_template_rules(self):
        for rule in ("`wf` is absent", "never create or edit TASKS.md", "Never ask questions",
                     "test red -> implement -> green", "commit on main", "network source is blocked",
                     "  WF-RESULT <done, awaiting or handback>\n", "  WF-REPORT <one or two lines",
                     'print("WF-USAGE in=%d cw=%d cr=%d out=%d model=%s"', 'echo "WF-PATCH-BEGIN sha256=$(',
                     "echo WF-PATCH-END", "base64 -w 76", "~/.claude/projects", "never -A", "__pycache__"):
            self.assertIn(rule, TEMPLATE)
        # live run 2026-10-06: '[WF-RESULT R]' placeholders made the session drop every key word
        self.assertNotIn("[WF-", TEMPLATE)
        self.assertIn("key word", TEMPLATE)

    def test_parser_never_matches_prompt(self):
        p = self.fill()
        for wrap in (False, True):
            self.assertIsNone(C.final_message(as_export(p, wrap)), wrap)

    def test_parser_takes_final_message_after_prompt(self):
        exp = as_export(self.fill(), True) + "\n● Working on it.\n  Ran 3 shell commands\n\n" + FINAL
        self.assertEqual(C.final_message(exp), ["WF-RESULT done", "WF-REPORT fixed it",
                                                "WF-USAGE in=1 cw=2 cr=3 out=4 model=m",
                                                "WF-PATCH-BEGIN sha256=ab bytes=3", "QUJD", "WF-PATCH-END"])

    def test_parser_last_result_wins_and_stops_at_next_message(self):
        exp = "● WF-RESULT handback\n  WF-REPORT old\n\n● WF-RESULT awaiting\n  WF-REPORT q?\n\n❯ more\n"
        self.assertEqual(C.final_message(exp), ["WF-RESULT awaiting", "WF-REPORT q?"])

    def test_include_problem(self):
        for bad in ("/abs", "../x", "a/../../b", "", ".git", ".git/config"):
            self.assertIsNotNone(C.include_problem(bad), bad)
        for ok in ("out/pack", "data.bin", "out/a/b/"):
            self.assertIsNone(C.include_problem(ok), ok)

    def test_size_refusal(self):
        self.assertIsNone(C.size_refusal(90_000_000))
        self.assertEqual(C.size_refusal(91_400_000), "snapshot 91 MB > 90 MB; trim cloud_include")


from test_claims import FOUR, git  # noqa: E402
from test_setup import GitCli  # noqa: E402

AREAS = "# Demo\n\n## Areas\n\n### app\n- Test recipe: RECIPE-MARK python3 -m unittest\n- Paths: src/app.py\n"



def stub_archive(tc, d: Path, status=200):
    """No network: wf_cloud.post records (url, headers) and answers `status`; fake credentials file."""
    creds = d / "credentials.json"
    creds.write_text('{"claudeAiOauth": {"accessToken": "tok-1"}}')
    tc.posts, tc.answer = [], status
    saved = (wf_cloud.post, wf_cloud.cli_version, os.environ.get("WF_CLAUDE_CREDENTIALS"))

    def post(url, headers, timeout=10):
        tc.posts.append((url, headers))
        if isinstance(tc.answer, Exception):
            raise tc.answer
        return tc.answer, "body"

    def restore():
        wf_cloud.post, wf_cloud.cli_version, env = saved
        os.environ.pop("WF_CLAUDE_CREDENTIALS", None)
        if env is not None:
            os.environ["WF_CLAUDE_CREDENTIALS"] = env
    wf_cloud.post, wf_cloud.cli_version = post, lambda: "9.9.9"
    os.environ["WF_CLAUDE_CREDENTIALS"] = str(creds)
    tc.addCleanup(restore)


class CloudProject(GitCli):
    """A cloud-opted git project + fake claude; wf cloud in-process."""
    tasks_text = FOUR.replace("Four.", "Four: fix src/app.py.")
    toml = GitCli.toml + 'cloud = true\ncloud_include = ["out/pack"]\ncloud_note = "NOTE-MARK data in out/pack"\n'

    def setUp(self):
        super().setUp()
        (self.root / "src").mkdir()
        (self.root / "src" / "app.py").write_text("v1\n")
        (self.root / "CLAUDE.md").write_text(AREAS)
        (self.root / "out").mkdir()
        (self.root / "out" / "tracked.txt").write_text("tracked under out/\n")
        git(self.root, "add", "-A")
        git(self.root, "add", "-f", "out/tracked.txt")
        git(self.root, "commit", "-qm", "app")
        (self.root / "src" / "app.py").write_text("dirty\n")          # working tree: not in the snapshot
        (self.root / "out" / "pack").mkdir(parents=True)
        (self.root / "out" / "pack" / "data.bin").write_text("pack\n")
        (self.root / "out" / "other.txt").write_text("not included\n")
        (self.root / ".wf").mkdir(exist_ok=True)
        (self.root / ".wf" / "x").write_text("state\n")
        self.d = Path(tempfile.mkdtemp())
        fake = self.d / "claude"
        fake.write_text(FAKE)
        fake.chmod(0o755)
        self.st = self.d / "state"
        self.old = (wf_cloud.CLAUDE, wf_cloud.CAP, dict(os.environ))
        wf_cloud.CLAUDE = str(fake)
        os.environ.update({"WF_CLAUDE_JSON": str(self.d / "claude.json"), "FAKE_STATE": str(self.d / "n"),
                           "FAKE_ARGS": str(self.d / "args"), "CLAUDE_CODE_MESSAGING_SOCKET": "",
                           "WF_INBOX": str(self.d / "inbox.md")})
        self.snap = self.root / "out" / "cloud" / "t-four"
        self.rec = self.root / ".wf" / "cloud" / "t-four.json"
        stub_archive(self, self.d)

    def tearDown(self):
        wf_cloud.CLAUDE, wf_cloud.CAP, env = self.old
        os.environ.clear()
        os.environ.update(env)
        shutil.rmtree(self.d)
        super().tearDown()

    def send(self, *extra):
        out, err = io.StringIO(), io.StringIO()
        with redirect_stdout(out), redirect_stderr(err):
            code = wf_cloud.main(["send", "t-four", "--project", str(self.root), *extra], state=self.st, now=T0)
        return code, out.getvalue(), err.getvalue()

    def ledger(self):
        return json.loads((self.st / "cloud.json").read_text()) if (self.st / "cloud.json").exists() else None

    def sh(self, cwd, *args):
        return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True).stdout.strip()

class Send(CloudProject):
    """wf cloud send against the fake claude."""

    def test_send_snapshot_prompt_record_claim(self):
        master = self.sh(self.root, "rev-parse", "master")
        code, out, err = self.send()
        self.assertEqual((code, err), (0, ""), out)
        sid = "session_01AbCdEfGhIjKlMnOpQrStUv"
        files = sorted(str(p.relative_to(self.snap)) for p in self.snap.rglob("*")
                       if p.is_file() and ".git" not in p.relative_to(self.snap).parts)
        self.assertEqual(files, [".gitignore", "CLAUDE.md", "DESIGN.md", "docs/plan.md", "out/pack/data.bin",
                                 "src/app.py", "workflow.toml"])
        self.assertEqual((self.snap / "src" / "app.py").read_text(), "v1\n")
        self.assertEqual(self.sh(self.snap, "log", "--format=%s"), f"base {master}")
        self.assertEqual(self.sh(self.snap, "ls-files").split("\n"), files)       # all committed (-f)
        self.assertEqual(self.sh(self.snap, "status", "--porcelain"), "")
        self.assertEqual(self.sh(self.snap, "branch", "--show-current"), "main")
        base = self.sh(self.snap, "rev-parse", "HEAD")
        cwd, prompt = (self.d / "args").read_text().split("\n", 1)
        self.assertEqual(Path(cwd), self.snap.resolve())
        self.assertIn("Task t-four:\n- **t-four** [P3] (1h): Four: fix src/app.py.\n", prompt)
        self.assertIn("RECIPE-MARK", prompt)
        self.assertIn("NOTE-MARK data in out/pack", prompt)
        self.assertIn(f"format-patch --binary {base}..HEAD", prompt)
        rec = json.loads(self.rec.read_text())
        self.assertEqual({k: rec[k] for k in ("id", "sid", "master", "base", "lane", "sent", "model")},
                         {"id": "t-four", "sid": sid, "master": master, "base": base, "lane": "slow",
                          "sent": "2026-10-06T12:00:00+00:00", "model": "claude-opus-5-5"})
        self.assertIn(f"**t-four** [P3] (1h) (in progress: cloud:{sid}): Four", self.tasks())
        led = self.ledger()
        self.assertEqual([(e["id"], e["sid"], e["state"]) for e in led["entries"]], [("t-four", sid, "running")])
        self.assertIn(f"sent t-four: {sid}", out)
        code, _, err = self.send()                                       # twice: refused
        self.assertEqual(code, 1)
        self.assertIn(f"t-four already sent ({sid})", err)

    def test_dry_run_sends_nothing(self):
        code, out, err = self.send("--dry-run")
        self.assertEqual((code, err), (0, ""))
        self.assertRegex(out, r"^snapshot 0\.0 MB \(master [0-9a-f]{12}, base [0-9a-f]{12}\)\n\nYou are a remote")
        self.assertIn("Task t-four:", out)
        self.assertFalse(self.snap.exists())
        self.assertFalse(self.snap.parent.exists())                      # out/cloud gone too, out/ kept
        self.assertFalse(self.rec.exists())
        self.assertIsNone(self.ledger())
        self.assertFalse((self.d / "n").exists())                         # claude never ran
        self.assertNotIn("in progress", self.tasks())

    def test_dry_run_keeps_live_snapshot(self):
        self.assertEqual(self.send()[0], 0)
        head = self.sh(self.snap, "rev-parse", "HEAD")
        code, out, err = self.send("--dry-run")
        self.assertEqual((code, err), (0, ""))
        self.assertEqual(self.sh(self.snap, "rev-parse", "HEAD"), head)    # live session's folder intact
        self.assertEqual(sorted(p.name for p in self.snap.parent.iterdir()), ["t-four"])                 # dry-run folder dropped

    def test_size_refused(self):
        wf_cloud.CAP = 10
        code, _, err = self.send()
        self.assertEqual(code, 1)
        self.assertIn("MB > 0 MB; trim cloud_include", err)
        self.assertFalse(self.snap.exists())
        self.assertIsNone(self.ledger())
        self.assertFalse((self.d / "n").exists())

    def test_ledger_refused_exit_3(self):
        with wf_cloud.locked(self.st) as led:
            for i in range(3):
                C.add(led, f"t{i}", "p", f"s{i}", "claude-opus-5-5", T0)
        code, out, err = self.send()
        self.assertEqual(code, 3)
        self.assertEqual(err.count("\n"), 1)
        self.assertIn("cloud refused: ", err)
        self.assertIn("max_parallel", err)
        self.assertEqual(len(self.ledger()["entries"]), 3)
        self.assertFalse(self.snap.exists())
        self.assertFalse(self.rec.exists())
        self.assertFalse((self.d / "n").exists())

    def test_send_failure_releases_reserve(self):
        os.environ["FAKE_MODE"] = "nosid"
        code, _, err = self.send()
        self.assertEqual(code, 1)
        self.assertIn("no session id", err)
        self.assertEqual(self.ledger()["entries"], [])
        self.assertFalse(self.snap.exists())
        self.assertFalse(self.rec.exists())

    def test_not_opted_in(self):
        (self.root / "workflow.toml").write_text(GitCli.toml)
        code, _, err = self.send()
        self.assertEqual(code, 1)
        self.assertIn("not opted in", err)

    def test_include_outside_refused(self):
        (self.root / "workflow.toml").write_text(GitCli.toml + 'cloud = true\ncloud_include = ["../x"]\n')
        code, _, err = self.send("--dry-run")
        self.assertEqual(code, 1)
        self.assertIn("cloud_include '../x': must be a path inside the project", err)


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


class PullPure(unittest.TestCase):
    def result(self, raw: bytes, sha=None, n=None):
        import base64, gzip, hashlib
        gz = gzip.compress(raw)
        b64 = base64.b64encode(gz).decode()
        return ["WF-RESULT done", "WF-REPORT fixed the parser,", "tests green",
                "WF-USAGE in=1000 cw=2000 cr=3000 out=4000 model=claude-opus-5-5",
                f"WF-PATCH-BEGIN sha256={sha or hashlib.sha256(gz).hexdigest()} bytes={n or len(gz)}",
                *[b64[i:i + 76] for i in range(0, len(b64), 76)], "WF-PATCH-END"]

    def test_parse_and_decode(self):
        r = C.parse_result(self.result(b"diff --git a/x b/x\n"))
        self.assertEqual((r.state, r.report, r.model), ("done", "fixed the parser, tests green", "claude-opus-5-5"))
        self.assertEqual((r.usage.inp, r.usage.cw, r.usage.cr, r.usage.out), (1000, 2000, 3000, 4000))
        self.assertEqual(C.decode_patch(r), b"diff --git a/x b/x\n")

    def test_header_wrapped_anywhere(self):
        lines = self.result(b"abc")
        head = lines[4]
        lines[4:5] = [head[:30], head[30:61], head[61:]]          # sha split over three lines
        self.assertEqual(C.decode_patch(C.parse_result(lines)), b"abc")

    def test_mismatches(self):
        with self.assertRaisesRegex(C.CloudError, "sha256 mismatch"):
            C.decode_patch(C.parse_result(self.result(b"abc", sha="0" * 64)))
        with self.assertRaisesRegex(C.CloudError, "bytes .* != 7 announced"):
            C.decode_patch(C.parse_result(self.result(b"abc", n=7)))
        with self.assertRaisesRegex(C.CloudError, "bad WF-RESULT 'maybe'"):
            C.parse_result(["WF-RESULT maybe"])
        with self.assertRaisesRegex(C.CloudError, "without WF-PATCH-END"):
            C.parse_result(self.result(b"abc")[:-1])
        self.assertIsNone(C.parse_result(["WF-RESULT handback", "WF-REPORT no"]).usage)

    def keyless(self, raw: bytes) -> str:
        """The keys-dropped shape seen live: '● done', report, usage, header split, base64, bare WF-PATCH-END."""
        import base64, gzip, hashlib
        gz = gzip.compress(raw)
        b64 = base64.b64encode(gz).decode()
        msg = ["done", "Added sub and a test.", "usage in=4 cw=5 cr=6 out=7 model=claude-opus-5-5",
               f"sha256={hashlib.sha256(gz).hexdigest()}", f"bytes={len(gz)}",
               *[b64[i:i + 76] for i in range(0, len(b64), 76)], "WF-PATCH-END"]
        return "\n".join(("● " if i == 0 else "  ") + ln for i, ln in enumerate(msg)) + "\n\n"

    def test_keyless_final_message(self):
        prompt = as_export("do it\nend with\n    [WF-PATCH-END]\nWF-PATCH-END", wrap=False)
        text = prompt + "● working\n\n" + self.keyless(b"diff --git a/x b/x\n") + "● Session resumed\n"
        self.assertIsNone(C.final_message(text))
        lines = C.keyless_message(text)
        self.assertEqual((lines[0], lines[-1]), ("done", "WF-PATCH-END"))
        r = C.keyless_result(lines)
        self.assertEqual((r.state, r.model, r.usage.inp, r.usage.out), ("handback", "claude-opus-5-5", 4, 7))
        self.assertEqual(C.decode_patch(r), b"diff --git a/x b/x\n")
        self.assertIsNone(C.keyless_message(prompt))                                   # prompt only: running
        self.assertIsNone(C.keyless_message(prompt + "● still working\n"))
        self.assertIsNone(C.keyless_message(text + as_export("redo with the keys", wrap=False)))   # redo sent
        self.assertIsNotNone(C.keyless_message(text + "❯ \n"))                       # empty input line
        r = C.keyless_result(["done", "no patch here", "WF-PATCH-END"])
        self.assertEqual((r.has_patch, r.usage), (False, None))

    def test_patch_files_and_problem(self):
        patch = ("diff --git a/src/a.py b/src/a.py\n--- a/src/a.py\n+++ b/src/a.py\n"
                 "diff --git a/old.txt b/new.txt\nrename from old.txt\nrename to new.txt\n")
        self.assertEqual(C.patch_files(patch), ["src/a.py", "old.txt", "new.txt"])
        never = ["TASKS.md", "tasks/archive.md", ".wf", "out", "data/pack"]
        self.assertIsNone(C.patch_problem(["src/a.py", "outish.txt"], "", never))
        self.assertIn("TASKS.md", C.patch_problem(["TASKS.md"], "", never))
        self.assertIn("data/pack", C.patch_problem(["data/pack/x"], "", never))
        self.assertIn("outside the tree", C.patch_problem(["../x"], "", never))
        self.assertIn("outside the tree", C.patch_problem([".git/hooks/x"], "", never))
        self.assertIn("outside the project folder", C.patch_problem(["other/x"], "proj", never))
        self.assertIsNone(C.patch_problem(["proj/src/a.py"], "proj", never))
        self.assertIn("out", C.patch_problem(["proj/out/x"], "proj", never))


class Pull(CloudProject):
    """wf cloud pull --export FIXTURE on a git project with a recorded send."""
    toml = None   # set in setUp: no worktree_setup (it writes log.txt), a quick_gate

    def setUp(self):
        from test_cli import TOML
        self.toml = (TOML + 'cloud = true\ncloud_include = ["out/pack"]\n'
                     'quick_gate = ["grep -q fixed src/app.py"]\n')
        super().setUp()
        os.environ.update({"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t",
                           "GIT_COMMITTER_EMAIL": "t@t"})
        (self.root / "src" / "app.py").write_text("v1\n")
        import wf
        from wflib import config
        cfg = config.load(self.root)
        self.master, self.base, _ = wf_cloud.snapshot(cfg, self.snap)
        self.sid = "session_01PullPullPullPull"
        with wf_cloud.locked(self.st) as led:
            C.add(led, "t-four", str(self.root), self.sid, C.MODEL, T0)
        self.rec.parent.mkdir(parents=True, exist_ok=True)
        self.rec.write_text(json.dumps({"id": "t-four", "sid": self.sid, "project": str(self.root), "lane": "slow",
                                        "master": self.master, "base": self.base, "folder": str(self.snap),
                                        "sent": T0.isoformat(), "model": C.MODEL, "bytes": 1}))
        with redirect_stdout(io.StringIO()):
            self.assertEqual(wf.main(["--project", str(self.root), "status", "t-four", "progress",
                                      f"cloud:{self.sid}"]), 0)
        self.exp = self.d / "export.txt"

    def change(self, files: dict[str, str]):
        """Commit files in the snapshot (the cloud's work) -> the gzip patch bytes."""
        import gzip
        for rel, text in files.items():
            (self.snap / rel).parent.mkdir(parents=True, exist_ok=True)
            (self.snap / rel).write_text(text)
        git(self.snap, "add", "-A", "-f")
        git(self.snap, "-c", "user.name=c", "-c", "user.email=c@x", "commit", "-qm", "cloud work")
        p = subprocess.run(["git", "-C", str(self.snap), "format-patch", "--binary", f"{self.base}..HEAD", "--stdout"],
                           capture_output=True, check=True).stdout
        return gzip.compress(p)

    def write_export(self, state="done", gz=b"", report="fixed app", sha=None, wrap=False, final=True):
        import base64, hashlib
        b64 = base64.b64encode(gz).decode()
        msg = [f"WF-RESULT {state}", f"WF-REPORT {report}", "WF-USAGE in=1000 cw=0 cr=0 out=1000 model=claude-opus-5-5",
               f"WF-PATCH-BEGIN sha256={sha or hashlib.sha256(gz).hexdigest()} bytes={len(gz)}",
               *[b64[i:i + 76] for i in range(0, len(b64), 76)], "WF-PATCH-END"]
        if wrap:   # the export soft-wraps at 60 columns: the header too
            msg = [part for ln in msg for part in textwrap.wrap(ln, 60, break_long_words=True)]
        text = as_export("prompt with ● WF-RESULT done inside\nWF-PATCH-BEGIN", wrap=False) + "\n"
        if final:
            text += "\n".join(("● " if i == 0 else "  ") + ln for i, ln in enumerate(msg)) + "\n\n❯ \n"
        self.exp.write_text(text)

    def pull(self, *extra, now=T0 + dt.timedelta(hours=1), export=True):
        out, err = io.StringIO(), io.StringIO()
        argv = ["pull", "t-four", "--project", str(self.root), "--no-push",
                *(["--export", str(self.exp)] if export else []), *extra]
        with redirect_stdout(out), redirect_stderr(err):
            code = wf_cloud.main(argv, state=self.st, now=now)
        return code, out.getvalue(), err.getvalue()

    def entry(self):
        return self.ledger()["entries"][0]

    def assert_ended(self, state):
        self.assertEqual(self.entry()["state"], state)
        self.assertFalse(self.snap.exists())
        self.assertFalse(self.rec.exists())

    def assert_handback(self, why, kept: bool):
        self.assert_ended("handback")
        self.assertIn(f"Recovery: cloud attempt {self.sid} — {why}", self.tasks())
        self.assertNotIn("in progress", self.tasks())
        self.assertEqual(self.sh(self.root, "rev-parse", "master"), self.master)
        self.assertEqual(self.sh(self.root, "branch", "--list", "slow/*"), "")
        self.assertEqual((self.root / "out" / "cloud" / "t-four.patch").exists(), kept)

    def test_done_archives_session(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n"}))
        code, out, err = self.pull()
        self.assertEqual((code, err), (0, ""), out)
        self.assertEqual(self.posts, [(f"https://api.anthropic.com/v1/code/sessions/{self.sid}/archive",
                                       {"Authorization": "Bearer tok-1", "Content-Type": "application/json",
                                        "anthropic-version": "2023-06-01", "User-Agent": "claude-code/9.9.9"})])
        self.assertIs(self.entry()["archived"], True)
        self.assertNotIn("archive", out.replace("archive.md", ""))

    def test_second_pull_same_id_skips(self):
        """Two pulls of one id at once (batch sidecar + orchestrator): the one without the record lock skips
        with rc 0, no 'branch … exists (local WIP?)'; the holder's pull still ends the task."""
        import fcntl
        self.write_export(gz=self.change({"src/app.py": "fixed\n"}))
        fd = os.open(self.rec, os.O_RDONLY)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)      # the other pull, mid-flight
            code, out, err = self.pull()
        finally:
            os.close(fd)
        self.assertEqual((code, err), (0, ""), out)
        self.assertEqual(out, "t-four: pulled by another wf cloud pull, skipped\n")
        self.assertTrue(self.rec.exists())
        self.assertEqual(self.entry()["state"], "running")
        self.assertEqual(self.sh(self.root, "branch", "--list", "slow/*"), "")
        code, out, err = self.pull()
        self.assertEqual((code, err), (0, ""), out)
        self.assert_ended("done")

    def test_pull_claim_record_gone(self):
        with wf_cloud.pull_claim(self.d / "gone.json") as mine:
            self.assertFalse(mine)
        with wf_cloud.pull_claim(self.rec) as mine:
            self.assertTrue(mine)
            self.rec.unlink()               # winner ended it while we waited: re-checked on the next claim
        with wf_cloud.pull_claim(self.rec) as mine:
            self.assertFalse(mine)

    def archive_fails(self, answer, why):
        self.answer = answer
        self.write_export(state="handback", report="cannot")
        code, out, err = self.pull()
        self.assertEqual((code, err), (0, ""), out)
        self.assert_ended("handback")
        self.assertIn(f"archive {self.sid} failed: {why}; archive by hand (wf cloud archive --ended)", out)
        self.assertNotIn("archived", self.entry())

    def test_archive_http_error_warns_pull_still_ends(self):
        self.archive_fails(403, "HTTP 403: body")

    def test_archive_network_error_warns_pull_still_ends(self):
        self.archive_fails(C.CloudError("timed out"), "timed out")

    def test_running_not_archived(self):
        self.write_export(final=False)
        code, out, err = self.pull()
        self.assertEqual(code, 4, out + err)
        self.assertEqual(self.posts, [])

    def test_archive_command_ended(self):
        self.write_export(state="handback", report="cannot")
        self.answer = 500
        self.pull()
        self.answer = 200
        out = io.StringIO()
        with redirect_stdout(out):
            code = wf_cloud.main(["archive", "--ended"], state=self.st, now=T0)
        self.assertEqual((code, out.getvalue()), (0, f"archived {self.sid}\n"))
        self.assertIs(self.entry()["archived"], True)
        with redirect_stdout(out := io.StringIO()):
            code = wf_cloud.main(["archive", "--ended"], state=self.st, now=T0)
        self.assertEqual((code, out.getvalue()), (0, "no ended cloud session left to archive\n"))

    def test_archive_command_one_sid_failure_exit_1(self):
        self.answer = 401
        err = io.StringIO()
        with redirect_stdout(io.StringIO()), redirect_stderr(err):
            code = wf_cloud.main(["archive", "session_01Other"], state=self.st, now=T0)
        self.assertEqual(code, 1)
        self.assertEqual(err.getvalue(), "wf: archive session_01Other failed: HTTP 401 (login expired? run claude once)\n")

    def test_done_applies_gates_merges(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n", "src/new.py": "new\n"}))
        code, out, err = self.pull()
        self.assertEqual((code, err), (0, ""), out)
        self.assertEqual(self.sh(self.root, "show", "master:src/app.py"), "fixed")
        self.assertEqual(self.sh(self.root, "show", "master:src/new.py"), "new")
        self.assertIn("cloud work", self.sh(self.root, "log", "--format=%s", "master"))
        self.assertNotIn("t-four", self.tasks())
        self.assertIn(f"fixed app (cloud {self.sid})", self.archive())
        self.assertEqual(self.sh(self.root, "status", "--porcelain", "TASKS.md", "tasks"), "")   # bookkeeping committed
        self.assert_ended("done")
        e = self.entry()
        self.assertEqual(e["usd_source"], "self")
        self.assertAlmostEqual(e["usd"], U.cost(C.MODEL, U.Usage(inp=1000, out=1000)) * C.OVERHEAD)
        self.assertFalse((self.root / "out" / "cloud" / "t-four.patch").exists())
        self.assertRegex(out, r"report: commit [0-9a-f]{7,}\n$")
        self.assertIn("$ grep -q fixed src/app.py", out)

    def test_wrapped_indented_base64_rejoined(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n"}), wrap=True)
        code, out, err = self.pull()
        self.assertEqual((code, err), (0, ""), out)
        self.assertEqual(self.sh(self.root, "show", "master:src/app.py"), "fixed")

    def test_sha_mismatch_hands_back(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n"}), sha="0" * 64)
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("patch sha256 mismatch", kept=False)

    def test_gate_red_hands_back_keeps_patch(self):
        self.write_export(gz=self.change({"src/app.py": "broken\n"}))
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("quick_gate 'grep -q fixed src/app.py' red (exit 1)", kept=True)
        wt = self.root / ".worktrees" / "slow"
        self.assertEqual(self.sh(wt, "status", "--porcelain"), "")
        self.assertEqual(self.sh(wt, "branch", "--show-current"), "")

    def test_patch_touching_cloud_include_refused(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n", "out/pack/data.bin": "x\n"}))
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("patch touches out/pack/data.bin", kept=True)

    def test_patch_touching_tasks_refused(self):
        self.write_export(gz=self.change({"src/app.py": "fixed\n", "TASKS.md": "x\n"}))
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("patch touches TASKS.md", kept=True)

    def test_awaiting_adds_question_and_blocks(self):
        self.write_export(state="awaiting", report="Which format: csv or json?")
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_ended("awaiting")
        m = __import__("re").search(r"\*\*(a-[^*]+)\*\*: Which format: csv or json\? \[\[t-four\]\]", self.tasks())
        self.assertTrue(m, self.tasks())
        self.assertIn(f"(blocked: [[{m[1]}]])", self.tasks())
        self.assertEqual(self.sh(self.root, "rev-parse", "master"), self.master)

    def test_cloud_handback(self):
        self.write_export(state="handback", report="needs the LAN host")
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("cloud handback: needs the LAN host", kept=False)

    def test_no_marker_running_exit_4(self):
        self.write_export(final=False)
        code, out, err = self.pull()
        self.assertEqual((code, err), (4, ""))
        self.assertIn(f"t-four: running ({self.sid}, sent 1h00m ago)", out)
        self.assertTrue(self.rec.exists() and self.snap.exists())
        self.assertEqual(self.entry()["state"], "running")
        self.assertIn(f"in progress: cloud:{self.sid}", self.tasks())

    def test_keyless_final_message_hands_back_keeps_patch(self):
        import gzip
        self.write_export(final=False)
        msg = PullPure().keyless(gzip.decompress(self.change({"src/app.py": "fixed\n"})))
        self.exp.write_text(self.exp.read_text() + msg + "● Session resumed\n")
        code, out, err = self.pull()
        self.assertEqual(code, 0, err)
        self.assert_handback("bad result: no WF-RESULT key; patch out/cloud/t-four.patch", kept=True)
        self.assertIn(b"+fixed", (self.root / "out" / "cloud" / "t-four.patch").read_bytes())
        self.assertEqual(self.entry()["usd_source"], "self")

    def test_teleport_export_used_and_removed(self):
        self.write_export(final=False)
        seen = []

        def fake_export(folder, sid, out, timeout=300):
            seen.append((Path(folder), sid))
            Path(out).parent.mkdir(parents=True, exist_ok=True)
            Path(out).write_text(self.exp.read_text())
            return Path(out)

        old, wf_cloud.export = wf_cloud.export, fake_export
        try:
            code, out, err = self.pull(export=False)
        finally:
            wf_cloud.export = old
        self.assertEqual((code, seen), (4, [(self.snap, self.sid)]))
        self.assertFalse((self.root / "out" / "cloud" / "t-four.export.txt").exists())

    def test_over_24h_lost(self):
        self.write_export(final=False)
        code, out, err = self.pull(now=T0 + dt.timedelta(hours=25))
        self.assertEqual(code, 0, err)
        self.assert_ended("lost")
        self.assertEqual((self.entry()["usd"], self.entry()["usd_source"]), (4.0, "est"))
        self.assertIn(f"Recovery: cloud attempt {self.sid} — lost: no WF-RESULT after 25h00m", self.tasks())
        self.assertNotIn("in progress", self.tasks())

    def test_all_and_not_sent(self):
        os.environ["FAKE_MODE"] = "noresume"      # teleport never resumes: export fails
        out, err = io.StringIO(), io.StringIO()
        with redirect_stdout(out), redirect_stderr(err):
            code = wf_cloud.main(["pull", "--all", "--project", str(self.root)], state=self.st,
                                 now=T0 + dt.timedelta(hours=25))
        self.assertEqual(code, 0, err.getvalue())          # no export, > 24h: lost
        self.assert_ended("lost")
        code, _, err = self.pull()
        self.assertEqual(code, 1)
        self.assertIn("t-four is not out in the cloud", err)


# ------------------------------------------------------------------ fit (spec §4.5)

FIT_DOC = """\
# Tasks — demo

## Awaiting your decision

## Pending

- **t-ok** [P1] (1h): Plain code. Goal.
  Done: tests pass
- **t-sonnet** [P1] (1h): Plain code. Goal.
  Done: tests pass
  Model: sonnet
- **t-slice** [P1] (5h): Big. Goal.
  Done: tests pass
- **t-owner** [P1] (1h): Plain. Goal.
  Done: tests pass
  Sessions: owner
- **t-solo** [P1] (1h): Plain. Goal.
  Done: tests pass
  Sessions: solo
- **t-nodone** [P1] (1h): Plain. Goal.
- **t-no** [P1] (1h): Plain. Goal.
  Done: tests pass
  Cloud: no
- **t-wf** [P1] (1h): Plain. Goal.
  Steps: run wf set x --model opus then check
  Done: tests pass
- **t-res** [P1] (1h): Plain. Goal.
  Steps: wf res run the build
  Done: tests pass
- **t-gui** [P1] (1h): Plain. Goal.
  Steps: click through the GUI
  Done: tests pass
- **t-lan** [P1] (1h): Plain. Goal.
  Steps: copy to 10.0.0.20
  Done: tests pass
- **t-srv** [P1] (1h): Plain. Goal.
  Steps: hit the live server
  Done: tests pass
- **t-yes-wf** [P1] (1h): Plain. Goal.
  Steps: run wf set x --model opus then check
  Done: tests pass
  Cloud: yes
- **t-yes-sonnet** [P1] (1h): Plain. Goal.
  Done: tests pass
  Model: sonnet
  Cloud: yes
- **t-yes-owner** [P1] (1h): Plain. Goal.
  Done: tests pass
  Sessions: owner
  Cloud: yes
- **t-yes-haiku** [P1] (<1h): Plain. Goal.
  Done: tests pass
  Model: haiku
  Cloud: yes
- **t-no-opus** [P1] (1h): Plain. Goal.
  Done: tests pass
  Model: opus
  Cloud: no
- **t-yes-slice** [P1] (5h): Big. Goal.
  Done: tests pass
  Cloud: yes

## Needs human

## Deferred
"""


class FitTest(unittest.TestCase):
    def setUp(self):
        from wflib import tasks as TK
        self.TK = TK
        self.doc = TK.parse(FIT_DOC)
        self.cfg = type("Cfg", (), {"cloud": True, "slice_above": "1h"})()

    def fit(self, id):
        return C.fit(self.doc.item(id), self.cfg)

    def test_fits(self):
        self.assertIsNone(self.fit("t-ok"))

    def test_exclusions(self):
        for id, why in [("t-sonnet", "Model sonnet"), ("t-slice", "slice"), ("t-owner", "not runner-ready"),
                        ("t-solo", "Sessions: solo"), ("t-nodone", "not runner-ready"), ("t-no", "Cloud: no"),
                        ("t-wf", "`wf` commands"), ("t-res", "wf res"), ("t-gui", "GUI"), ("t-lan", "LAN"),
                        ("t-srv", "live server")]:
            with self.subTest(id):
                self.assertIn(why, self.fit(id) or "")

    def test_not_opted_in(self):
        self.cfg.cloud = False
        self.assertIn("not opted in", self.fit("t-ok"))

    def test_cloud_yes_skips_regex_only_opus_only(self):
        self.assertIsNone(self.fit("t-yes-wf"))
        self.assertEqual(self.fit("t-yes-sonnet"), "Model sonnet stays local")
        self.assertEqual(self.fit("t-yes-haiku"), "Model haiku stays local")
        self.assertEqual(self.fit("t-no-opus"), "Cloud: no")
        self.assertEqual(self.fit("t-sonnet"), "Model sonnet stays local")    # unset: old rule
        self.assertIn("not runner-ready", self.fit("t-yes-owner"))
        self.assertIn("slice", self.fit("t-yes-slice"))

    def test_cloud_property_and_set(self):
        self.assertEqual(self.doc.item("t-no").cloud, "no")
        self.assertIsNone(self.doc.item("t-ok").cloud)
        self.TK.set_fields(self.doc, "t-ok", cloud="yes")
        self.assertEqual(self.doc.item("t-ok").body, ["  Done: tests pass", "  Cloud: yes"])
        self.TK.set_fields(self.doc, "t-ok", cloud="no", model="sonnet")
        self.assertEqual(self.doc.item("t-ok").body, ["  Done: tests pass", "  Model: sonnet", "  Cloud: no"])
        self.TK.set_fields(self.doc, "t-ok", cloud="")
        self.assertEqual(self.doc.item("t-ok").body, ["  Done: tests pass", "  Model: sonnet"])
        with self.assertRaises(self.TK.TaskError):
            self.TK.set_fields(self.doc, "t-ok", cloud="maybe")


class PickKeyTest(unittest.TestCase):
    def test_order(self):
        from wflib import tasks as TK
        doc = TK.parse("""# Tasks

## Pending

- **t-p1-small** [P1] (<1h): A. G.
  Done: x
- **t-p1-big** [P1] (1h): A. G.
  Done: x
- **t-p2-yes** [P2] (<1h): A. G.
  Done: x
  Cloud: yes
- **t-p0** [P0] (<1h): A. G.
  Done: x

## Needs human

## Deferred
""")
        items = [doc.item(i) for i in ("t-p1-small", "t-p1-big", "t-p2-yes", "t-p0")]
        got = [i.id for i in sorted(items, key=C.pick_key)]
        # yes first; then prio; then 1h before <1h
        self.assertEqual(got, ["t-p2-yes", "t-p0", "t-p1-big", "t-p1-small"])


class CloudLineCheck(unittest.TestCase):
    def test_check(self):
        sys.path.insert(0, str(HERE / "tests"))
        from test_check import Base

        class B(Base):
            def runTest(self):
                self.pending("- **t-a** [P1] (1h): A. G.\n  Done: x\n  Cloud: yes\n")
                self.assertEqual(self.run_check(), ([], []))
                self.pending("- **t-a** [P1] (1h): A. G.\n  Done: x\n  Cloud: maybe\n")
                self.assertTrue(any("Cloud 'maybe'" in e for e in self.errors()))
                self.pending("- **t-a** [P1] (1h): A. G.\n  Done: x\n  Cloud: yes\n  Cloud: no\n")
                self.assertTrue(any("two Cloud lines" in e for e in self.errors()))
        r = unittest.TextTestRunner(stream=io.StringIO()).run(B())
        self.assertTrue(r.wasSuccessful(), r.failures + r.errors)


ORCH_TASKS = """\
# Tasks — demo

## Awaiting your decision

## Pending

- **t-fast** [P1] (<1h): Fast sonnet task.
  - Done: fast works
  - Model: sonnet

- **t-gui** [P1] (<1h): Check the GUI screenshot.
  - Done: gui works

- **t-hai** [P1] (<1h): Haiku forced to the cloud.
  - Done: hai works
  - Model: haiku
  - Cloud: yes

- **t-slow** [P2] (1h): Slow opus task.
  - Done: slow works

- **t-yes** [P3] (<1h): Opus task marked for the cloud.
  - Done: yes works
  - Cloud: yes

## Needs human

## Deferred
"""


class OrchCloud(CloudProject):
    """wf orch pick cloud (subprocess, fake claude via WF_CLAUDE, ledger via XDG_STATE_HOME)."""
    tasks_text = ORCH_TASKS

    def setUp(self):
        super().setUp()
        from test_merge import IDENT
        self.env = {**IDENT, "WF_CLAUDE": wf_cloud.CLAUDE, "XDG_STATE_HOME": str(self.d / "xdg")}

    def orch(self, *args):
        out = self.ok("orch", *args, env=self.env)
        return out

    def put_ledger(self, **kw):
        n = kw.pop("running", 0)
        led = {**C.new(), **kw}
        for i in range(n):
            C.add(led, f"t-x{i}", "/x", f"session_x{i}", "opus", T0)
        (self.d / "xdg" / "wf").mkdir(parents=True, exist_ok=True)
        (self.d / "xdg" / "wf" / "cloud.json").write_text(C.dumps(led))

    def status(self, id):
        return next(l for l in (self.root / "TASKS.md").read_text().splitlines() if f"**{id}**" in l)

    def test_pick_cloud_yes_first_then_opus_then_none(self):
        out = self.orch("pick", "cloud")
        self.assertIn("pick: t-yes (lane cloud, model opus", out)      # Cloud: yes beats P2 1h; haiku never
        out = self.orch("pick", "cloud")
        self.assertIn("pick: t-slow (lane cloud, model opus", out)
        self.assertIn("in progress: cloud:session_01AbCdEfGhIjKlMnOpQrStUv", self.status("t-slow"))
        rec = json.loads((self.root / ".wf" / "orch" / "t-slow.json").read_text())
        self.assertEqual((rec["lane"], rec["task_lane"], rec["branch"]), ("cloud", "slow", "slow/t-slow"))
        self.assertTrue((self.root / ".wf" / "cloud" / "t-slow.json").exists())
        out = self.orch("pick", "cloud")
        self.assertEqual(out.strip().splitlines()[0][:26], "stop lane cloud: none fit ")
        led = json.loads((self.d / "xdg" / "wf" / "cloud.json").read_text())
        self.assertEqual([e["id"] for e in led["entries"]], ["t-yes", "t-slow"])

    def test_local_lane_skips_cloud_claim(self):
        self.orch("pick", "cloud", "--id", "t-slow")
        out = self.orch("pick", "slow")
        self.assertNotIn("t-slow", out.splitlines()[0])
        self.assertIn("pick: t-fast", out)

    def test_stop_ledger(self):
        self.put_ledger(budget=3.0)
        out = self.orch("pick", "cloud")
        self.assertIn("stop lane cloud: ledger (balance $3.00 < reserve $4.00)", out)
        self.assertNotIn("in progress", self.status("t-slow"))

    def test_stop_max_parallel(self):
        self.put_ledger(max_parallel=2, running=2)
        out = self.orch("pick", "cloud")
        self.assertIn("stop lane cloud: max parallel (2 running >= max_parallel 2)", out)
        self.assertFalse((self.root / ".wf" / "cloud" / "t-slow.json").exists())

    def test_id_unfit_refused(self):
        out = self.orch("pick", "cloud", "--id", "t-gui")
        self.assertIn("stop lane cloud: none fit (t-gui: needs a GUI)", out)

    def test_not_opted_in(self):
        (self.root / "workflow.toml").write_text(GitCli.toml)
        out = self.orch("pick", "cloud")
        self.assertIn("stop lane cloud: none fit (project not opted in", out)

    def test_post_handback_commits_and_stops(self):
        self.orch("pick", "cloud", "--id", "t-slow")
        self.ok("note", "t-slow", "Recovery: cloud attempt x — red")       # what wf cloud pull leaves behind
        self.ok("status", "t-slow", "clear")
        out = self.orch("post", "t-slow", "cloud", "--result", "handback")
        self.assertIn("post: t-slow handback", out)
        self.assertIn("committed leftover", out)
        self.assertIn("stop lane cloud: handback", out)
        self.assertIn(" cloud opus t-slow handback ", (self.root / "out" / "wf-orch.log").read_text())