"""A workflow agent's five-hour limit moves a claude session: actual tmux, shell, production
shim and watcher, with the cross-provider suite's disposable fake Claude TUI."""

import json
from pathlib import Path
import signal
import time
import unittest

from autoresume_cross_tmux_support import CrossScenario, TMUX

AGENT = "a33659a7131e55067"
AGENT2 = "ab12cd34ef56a7890"


def stamp(now):
    """A transcript timestamp to the millisecond, as claude 2.1.303 writes them."""
    return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(now)) + ".%03dZ" % int(now % 1 * 1000)


@unittest.skipUnless(TMUX, "real tmux is unavailable")
class SidechainTmuxTests(unittest.TestCase):
    def test_a_workflow_agents_limit_resumes_the_idle_session_on_another_account(self):
        # The suite's short grace (0.4 s): the relaunch starts a second or two after the
        # rejection, and one more agent is rejected as the client is stopped (the fake TUI
        # records it on SIGTERM). The relaunched watcher reads both in the same agent
        # transcripts and must act on neither: they are the replaced process's.
        case = CrossScenario(CLAUDE_MULTIACC_AR_SUBAGENT_SCAN="0.2",
                             FAKE_AGENT_LIMIT_ON_TERM=AGENT2)
        self.addCleanup(case.close)
        case.start("claude")
        ready = case.records("ready")[0]
        # this launch's own agents start a second after it at the soonest
        state = case.pools["claude"] / "tmp" / "autoresume" / ("%d.state" % ready["pid"])
        launched = int(dict(line.split("=", 1) for line in state.read_text().splitlines()
                            if "=" in line)["launched"])
        time.sleep(max(0.0, launched + 1.05 - time.time()))
        # the main transcript stays quiet; only the workflow's agent records the limit
        agent = (Path(ready["path"]).with_suffix("") / "subagents" / "workflows" / "wf_e2e"
                 / ("agent-%s.jsonl" % AGENT))
        agent.parent.mkdir(parents=True)
        reset = int(time.time()) + 3600
        agent.write_text(json.dumps({
            "type": "assistant", "isSidechain": True, "agentId": AGENT, "sessionId": ready["sid"],
            "timestamp": stamp(time.time()),
            "isApiErrorMessage": True, "error": "rate_limit", "apiError": "usage_limit_reached",
            "apiErrorStatus": 429, "quotaLimits": {"status": "rejected", "resetsAt": reset,
                                                   "rateLimitType": "five_hour",
                                                   "overageStatus": "rejected"},
            "message": {"model": "<synthetic>", "role": "assistant", "content": [
                {"type": "text", "text": "You've hit your session limit"}]}}, separators=(",", ":"))
            + "\n")
        case.wait(lambda: len(case.records("ready")) >= 2, "same session resumed on another account")
        case.wait(lambda: "autoresume switch" in case.logs("claude"), "watcher confirms switch")
        first, second = case.records("launch")[:2]
        self.assertEqual([first["provider"], second["provider"]], ["claude", "claude"])
        self.assertEqual((first["account"], second["account"]), ("acct-01", "acct-02"))
        self.assertEqual(first["sid"], second["sid"])
        self.assertEqual(second["argv"][second["argv"].index("--resume") + 1], first["sid"])
        self.assertIn("Continue", second["argv"][-1])
        self.assertIn("class=quota code=rate_limit sid=%s acct=acct-01 reset=%d src=sidechain agent=%s"
                      % (first["sid"], reset, AGENT), case.logs("claude"))
        self.assertEqual(case.records("keys"), [], "no keystroke may reach either TUI")
        self.assertEqual(case.records("signal")[0]["signal"], signal.SIGTERM)
        self.assertEqual([row["agent"] for row in case.records("agent-limit")], [AGENT2])
        marker = (case.pools["claude"] / "acct-01" / ".limited").read_text().split("\n")
        self.assertEqual(marker[0], str(reset))
        self.assertIn("bucket=client:five_hour", marker[1])
        # the relaunched session's watcher reads the same agent transcripts, and lets them be
        case.wait(lambda: case.logs("claude").count("autoresume watch") == 2, "second watcher")
        time.sleep(1)
        self.assertEqual(case.logs("claude").count("autoresume detect"), 1)
        self.assertEqual(len(case.records("launch")), 2)
        self.assertFalse((case.pools["claude"] / "acct-02" / ".limited").exists(),
                         "the account the session moved to is never parked")


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