skills/tests/test_rasci.py

"""Exercise the kvb-org-rasci register: fields, roles, priority order, claims, artifacts, tools."""

from __future__ import annotations

import datetime
import json
import os
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
RASCI = REPO_ROOT / "skills" / "kvb-org-rasci" / "scripts" / "rasci.py"
COUNCIL = REPO_ROOT / "skills" / "kvb-org-organize" / "scripts" / "council.py"
WIDE_HEADER = "| Prio | WSJF | Task | OKR | Status | Worker | Artifacts | Start | End | R | A | S | C | I |"
COMPACT_HEADER = "| Prio | WSJF | Task | Status | Worker | A |"


def local(iso: str) -> str:
    return datetime.datetime.fromisoformat(iso).astimezone().strftime("%m-%d %H:%M")


class TestRasci(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.home = Path(self.tmp.name) / "home"
        self.root = Path(self.tmp.name) / "council"
        self.sessions = self.home / ".claude" / "sessions"
        self.sessions.mkdir(parents=True)
        for sid, name in (("s1", "One"), ("s2", "Two")):
            (self.sessions / f"{sid}.json").write_text(
                json.dumps({"sessionId": sid, "name": name}), encoding="utf-8"
            )

    def _run(self, *args: str, script: Path = RASCI, sid: str = "s1", timeout: float = 120) -> subprocess.CompletedProcess:
        env = {
            "PATH": "/usr/bin:/bin",
            "HOME": str(self.home),
            "AGENTS_COUNCIL_ROOT": str(self.root),
            "CLAUDE_CODE_SESSION_ID": sid,
        }
        return subprocess.run(
            [sys.executable, str(script), *args],
            capture_output=True,
            text=True,
            env=env,
            cwd=self.tmp.name,
            check=False,
            timeout=timeout,
        )

    def _out(self, *args: str) -> list[str]:
        result = self._run(*args)
        self.assertEqual(result.returncode, 0, result.stderr)
        return result.stdout.splitlines()

    def _claim(self, office: str, session: str, name: str, work: str) -> None:
        path = self.root / office
        path.mkdir(parents=True)
        claim = {
            "office": office,
            "session": session,
            "runtime": "claude-code",
            "name": name,
            "since": "2026-11-01T00:00:00+00:00",
            "beat": time.time(),
            "work": work,
        }
        (path / "claim.json").write_text(json.dumps(claim), encoding="utf-8")

    def _two_tasks(self) -> str:
        start = "2026-11-01T10:00:00+00:00"
        self._run("set", "work/a", "--title", "Alpha", "--objective", "Ship A", "--prio", "2",
                  "--status", "running", "--worker", "Skills", "--start", start)
        self._run("role", "work/a", "Skills", "ar")
        self._run("role", "work/a", "Operator", "ic")
        self._run("role", "work/a", "Pkgmgr", "C")
        self._run("set", "work/b", "--title", "Beta", "--prio", "1", "--worker", "subagent Explore (haiku)")
        self._run("role", "work/b", "Pkgmgr", "R")
        return start

    def test_wide_matrix_orders_by_priority_and_lists_roles(self):
        start = self._two_tasks()
        lines = self._out("show", "--wide")
        self.assertEqual(lines[0], WIDE_HEADER)
        self.assertEqual(
            lines[2], "| 1 | - | work/b: Beta | - | queued | subagent Explore (haiku) | - | - | - | Pkgmgr |  |  |  |  |"
        )
        self.assertEqual(
            lines[3],
            f"| 2 | - | work/a: Alpha | Ship A | running | Skills | - | {local(start)} | - "
            "| Skills | Skills |  | Operator, Pkgmgr | Operator |",
        )
        self.assertEqual(
            lines[4], "gaps: work/b has 0 A; work/b has no key result; work/a has no key result"
        )

    def test_show_prints_the_compact_table_by_default(self):
        self._two_tasks()
        self.assertEqual(
            self._out("show"),
            [
                COMPACT_HEADER,
                "|---|---|---|---|---|---|",
                "| 1 | - | work/b: Beta | queued | subagent Explore (haiku) |  |",
                "| 2 | - | work/a: Alpha | running | Skills | Skills |",
                "gaps: work/b has 0 A; work/b has no key result; work/a has no key result",
            ],
        )

    def test_compact_cells_keep_the_slug_and_truncate_title_worker_and_a(self):
        slug = "work/" + "s" * 45
        self._run("set", slug, "--title", "Hidden", "--prio", "1")
        self._run("set", "work/a", "--title", "A title far longer than the forty character cell",
                  "--prio", "2", "--worker", "subagent general-purpose (default)")
        self._run("role", "work/a", "Accountable Party With A Long Name", "A")
        lines = self._out("show")
        self.assertEqual(lines[2], f"| 1 | - | {slug}… | queued | - |  |")
        self.assertEqual(
            lines[3],
            "| 2 | - | work/a: A title far longer than the for… | queued | subagent general-purpos… "
            "| Accountable Party With … |",
        )

    def test_key_results_gate_done_and_fill_the_okr_column(self):
        self._run("set", "work/a", "--objective", "Ship A", "--status", "running")
        self._run("role", "work/a", "Skills", "RA")
        self._run("kr", "work/a", "add", "tests green")
        self._run("kr", "work/a", "add", "released")
        self._run("kr", "work/a", "met", "1")
        lines = self._out("show", "--wide")
        self.assertIn("| Ship A (1/2 KR) | running |", lines[2])
        self.assertEqual(len(lines), 3)
        self.assertNotIn("Ship A", self._out("show")[2])
        self.assertEqual(self._run("set", "work/a", "--status", "done").returncode, 2)
        self.assertIn("key result 2 [ ]: released", self._out("get", "work/a"))
        self._run("kr", "work/a", "met", "2")
        self.assertEqual(self._run("set", "work/a", "--status", "done").returncode, 0)
        self.assertEqual(self._out("show"), ["no tasks"])

    def test_status_sets_start_and_end(self):
        self._run("set", "work/a", "--status", "running")
        self._run("set", "work/a", "--status", "done", "--description", "long text")
        lines = self._out("get", "work/a")
        self.assertIn("status: done", lines)
        self.assertIn("description: long text", lines)
        self.assertTrue(any(line.startswith("start: 20") for line in lines))
        self.assertTrue(any(line.startswith("end: 20") for line in lines))
        self._run("set", "work/a", "--status", "running")
        self.assertFalse(any(line.startswith("end:") for line in self._out("get", "work/a")))

    def test_base_names_the_upstream_branch_after_the_branch(self):
        self._run("set", "work/side", "--branch", "sidetree", "--base", "line")
        lines = self._out("get", "work/side")
        self.assertIn("base: line", lines)
        self.assertEqual(lines.index("base: line"), lines.index("branch: sidetree") + 1)
        self.assertEqual(json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/side"]["base"], "line")
        self._run("set", "work/side", "--base", "")
        self.assertFalse(any(line.startswith("base:") for line in self._out("get", "work/side")))

    def _after(self) -> dict[str, list[str]]:
        rows = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))
        return {task: row["after"] for task, row in rows.items() if "after" in row}

    def test_after_stores_and_drops_a_dependency_and_refuses_unknown_tasks_and_cycles(self):
        for task in ("work/a", "work/b", "work/c"):
            self._out("set", task)
        self._out("set", "work/b", "--after", "work/a")
        self._out("set", "work/c", "--after", "work/b", "--after", "work/a", "--after", "work/b")
        self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/b", "work/a"]})
        self.assertIn("after: work/b", self._out("get", "work/c"))
        before = (self.root / "rasci.json").read_bytes()
        for args, reason in (
            (("set", "work/a", "--after", "work/c"), "closes a cycle"),
            (("set", "work/a", "--after", "work/b", "--title", "kept out"), "closes a cycle"),
            (("set", "work/a", "--after", "work/a"), "cannot follow itself"),
            (("set", "work/a", "--after", "work/nope"), "no task of the register"),
            (("set", "work/new", "--after", "work/nope"), "no task of the register"),
            (("set", "work/a", "--after", "Bad Task"), "invalid task"),
        ):
            with self.subTest(args=args):
                result = self._run(*args)
                self.assertEqual(result.returncode, 2, result.stdout)
                self.assertIn(reason, result.stderr)
                self.assertEqual((self.root / "rasci.json").read_bytes(), before)
        self._out("set", "work/c", "--drop-after", "work/b")
        self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/a"]})
        self._out("set", "work/b", "--drop-after", "work/a")
        self._out("set", "work/a", "--after", "work/b")
        self.assertEqual(self._after(), {"work/a": ["work/b"], "work/c": ["work/a"]})

    def test_the_operator_business_value_replaces_the_median_and_may_stand_alone(self):
        self._run("estimate", "work/a", "One", "--value", "8", "--complexity", "2")
        self._run("estimate", "work/a", "Two", "--value", "13", "--complexity", "3")
        self._out("estimate", "work/a", "Operator", "--value", "34")
        row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"]
        self.assertNotIn("complexity", row["estimates"]["Operator"])
        self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 34 | 2.5 | 13.6 | One, Operator, Two | queued |")
        self.assertIn("estimate Operator: business value 34, complexity -", self._out("get", "work/a"))
        self._out("estimate", "work/a", "Operator", "--value", "21", "--complexity", "8")
        self._out("estimate", "work/a", "Operator", "--value", "5")
        row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"]
        self.assertEqual((row["estimates"]["Operator"]["value"], row["estimates"]["Operator"]["complexity"]), (5, 8))
        self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 5 | 3 | 1.67 | One, Operator, Two | queued |")
        self._out("estimate", "work/b", "Operator", "--value", "8")
        self.assertEqual(self._out("wsjf")[3:], ["| 2 | work/b | - | - | - | - | Operator | queued |", "unestimated: work/b"])

    def test_done_tasks_show_only_with_all(self):
        self._run("set", "work/a", "--status", "done", "--end", "2026-11-01T12:00:00+00:00")
        self._run("role", "work/a", "Skills", "RA")
        self.assertEqual(self._out("show"), ["no tasks"])
        self.assertEqual(self._out("show", "--wide"), ["no tasks"])
        self.assertEqual(self._out("show", "--all")[2], "| 1 | - | work/a | done | - | Skills |")
        self.assertIn(
            f"| done | - | - | - | {local('2026-11-01T12:00:00+00:00')} |", self._out("show", "--all", "--wide")[2]
        )

    def test_artifacts_and_tools(self):
        self._run("set", "work/a", "--workdir", "/w/a", "--branch", "feat/a", "--status", "done",
                  "--artifact", "image:core:dev", "--artifact", "volume:db",
                  "--tool", "council.py", "--tool", "make test")
        self._run("set", "work/b", "--worker", "Two", "--tool", "make test", "--artifact", "dir:/tmp/b")
        self._run("set", "work/a", "--drop-artifact", "volume:db")
        self._run("set", "work/b", "--branch", "")
        rows = [line.split("\t") for line in self._out("artifacts")]
        self.assertEqual(
            [row[:4] for row in rows],
            [
                ["done", "work/a", "-", "workdir:/w/a"],
                ["done", "work/a", "-", "branch:feat/a"],
                ["done", "work/a", "-", "image:core:dev"],
                ["queued", "work/b", "Two", "dir:/tmp/b"],
            ],
        )
        self.assertEqual(rows[0][4], "-")
        self.assertRegex(rows[2][4], r"^\d\d-\d\d \d\d:\d\d$")
        self.assertEqual([r.split("\t")[1] for r in self._out("artifacts", "--ended")], ["work/a"] * 3)
        self.assertEqual(len(self._out("artifacts", "--task", "work/b")), 1)
        self.assertIn("| queued | Two | 1 |", self._out("show", "--wide")[2])
        self.assertTrue(any(line.startswith("artifact: image:core:dev (since ") for line in self._out("get", "work/a")))
        self.assertEqual(
            self._out("tools"), ["2\tmake test\twork/a work/b", "1\tcouncil.py\twork/a"]
        )
        self.assertIn("tool: council.py", self._out("get", "work/a"))

    def test_claims_join_the_matrix_with_holder_as_ra(self):
        self._claim("work/live", "s1", "One", "Gamma")
        self._claim("cluster/dead", "s9", "Gone", "Delta")
        self._claim("chair", "s1", "One", "allocator")
        self._run("set", "work/live", "--prio", "1", "--status", "running")
        self._run("role", "work/live", "Two", "S")
        lines = self._out("show", "--wide")
        self.assertTrue(lines[2].startswith("| 1 | - | work/live | - | running | One |"))
        self.assertTrue(lines[2].endswith("| One | One | Two |  |  |"))
        self.assertTrue(lines[3].startswith("| 2 | - | cluster/dead | - | stale | Gone |"))
        self.assertFalse(any("chair" in line for line in lines))
        lines = self._out("show")
        self.assertEqual(lines[2], "| 1 | - | work/live | running | One | One |")
        self.assertEqual(lines[3], "| 2 | - | cluster/dead | stale | Gone | Gone |")
        self.assertFalse(any("chair" in line for line in lines))

    def test_council_claim_takeover_and_release_write_through(self):
        self.assertEqual(self._run("claim", "work/x", "--work", "Do X", script=COUNCIL).returncode, 0)
        lines = self._out("get", "work/x")
        self.assertIn("status: running", lines)
        self.assertIn("worker: One", lines)
        self.assertIn("description: Do X", lines)
        self.assertIn(f"workdir: {self.tmp.name}", lines)
        self.assertIn("roles: One=RA", lines)
        (self.sessions / "s1.json").unlink()
        self.assertEqual(self._run("takeover", "work/x", "--work", "Do X", script=COUNCIL, sid="s2").returncode, 0)
        lines = self._out("get", "work/x")
        self.assertIn("worker: Two", lines)
        self.assertIn("roles: Two=RA", lines)
        self._run("kr", "work/x", "add", "X shipped")
        self.assertEqual(self._run("release", "work/x", script=COUNCIL, sid="s2").returncode, 0)
        lines = self._out("get", "work/x")
        self.assertIn("status: stopped", lines)
        self.assertTrue(any(line.startswith("end: 20") for line in lines))

    def test_rm_and_clear_a_party(self):
        self._run("set", "work/a", "--title", "Alpha")
        self._run("role", "work/a", "Skills", "RA")
        self._run("role", "work/a", "Pkgmgr", "C")
        self._run("role", "work/a", "Pkgmgr", "-")
        self.assertTrue(self._out("show", "--wide")[2].endswith("| Skills | Skills |  |  |  |"))
        self.assertEqual(self._out("rm", "work/a"), ["removed work/a"])
        self.assertEqual(self._out("show"), ["no tasks"])

    def test_invalid_input_exits_2(self):
        for args in (
            ("role", "work/a", "Skills", "RR"),
            ("role", "work/a", "Skills", "Q"),
            ("set", "Bad Task"),
            ("set", "work/a", "--prio", "0"),
            ("set", "work/a", "--status", "finished"),
            ("set", "work/a", "--start", "yesterday"),
            ("get", "work/none"),
            ("kr", "work/a", "met", "1"),
            ("set", "work/a", "--artifact", "no-kind"),
            ("estimate", "work/a", "One", "--value", "4", "--complexity", "2"),
            ("estimate", "work/a", "One", "--value", "8"),
            ("estimate", "work/a", "Operator", "--complexity", "8"),
            ("estimate", "work/a", "Operator", "--value", "4"),
        ):
            with self.subTest(args=args):
                self.assertEqual(self._run(*args).returncode, 2)
        self.assertFalse((self.root / "rasci.json").exists())

    def test_pipe_in_a_title_does_not_break_the_table(self):
        self._run("set", "work/a", "--title", "a | b")
        self._run("role", "work/a", "Skills", "RA")
        self.assertTrue(self._out("show", "--wide")[2].startswith("| 1 | - | work/a: a \\| b | - | queued |"))
        self.assertEqual(self._out("show")[2], "| 1 | - | work/a: a \\| b | queued | - | Skills |")

    def test_wsjf_ranks_by_median_estimates(self):
        self._run("set", "work/a", "--objective", "Ship A")
        self._run("kr", "work/a", "add", "released")
        self._run("estimate", "work/a", "One", "--value", "8", "--complexity", "2")
        self._run("estimate", "work/a", "Two", "--value", "13", "--complexity", "3")
        self._run("estimate", "work/b", "One", "--value", "5", "--complexity", "5")
        self._run("set", "work/c", "--title", "Gamma")
        lines = self._out("wsjf")
        self.assertEqual(
            lines[0], "| Rank | Task | OKR | Business value | Complexity | WSJF | Estimators | Status |"
        )
        self.assertEqual(lines[2], "| 1 | work/a | Ship A (0/1 KR) | 10.5 | 2.5 | 4.2 | One, Two | queued |")
        self.assertEqual(lines[3], "| 2 | work/b | - | 5 | 5 | 1 | One | queued |")
        self.assertEqual(lines[4], "| 3 | work/c | - | - | - | - | - | queued |")
        self.assertEqual(lines[5], "unestimated: work/c")
        self.assertTrue(self._out("show")[2].startswith("| 1 | 4.2 | work/a |"))
        self.assertIn("wsjf: 4.2", self._out("get", "work/a"))
        self._run("set", "work/c", "--prio", "1")
        self.assertTrue(self._out("show")[2].startswith("| 1 | - | work/c: Gamma |"))
        self._run("estimate", "work/b", "One", "--drop")
        self.assertFalse(any(line.startswith("estimate ") for line in self._out("get", "work/b")))

    def test_hotfix_and_operator_rows_are_no_work_tasks_for_gaps_and_wsjf(self):
        self._run("set", "work/a")
        self._run("role", "work/a", "Skills", "RA")
        self._run("kr", "work/a", "add", "released")
        self._run("set", "hotfix/db", "--status", "running")
        self._run("set", "operator/push", "--status", "running")
        shown = self._out("show")
        self.assertEqual([line.split(" | ")[2] for line in shown[2:]], ["hotfix/db", "operator/push", "work/a"])
        self.assertFalse(any(line.startswith("gaps:") for line in shown))
        self.assertEqual(
            self._out("wsjf")[2:], ["| 1 | work/a | - (0/1 KR) | - | - | - | - | queued |", "unestimated: work/a"]
        )
        self._run("set", "work/b", "--title", "Beta")
        self.assertEqual(self._out("show")[-1], "gaps: work/b has 0 A; work/b has no R; work/b has no key result")
        self._run("set", "hotfix/db", "--status", "done", "--branch", "fix/db", "--workdir", "/w/db")
        self._run("set", "operator/push", "--branch", "main", "--artifact", "dir:/tmp/push")
        self._run("set", "work/a", "--branch", "feat/a")
        for args in (("artifacts",), ("artifacts", "--ended"), ("artifacts", "--task", "hotfix/db")):
            with self.subTest(args=args):
                self.assertEqual([line.split("\t")[1] for line in self._out(*args)],
                                 ["work/a"] if args == ("artifacts",) else [])

    def _ranked(self) -> list[str]:
        return [line.split(" | ")[2].split(":")[0] for line in self._out("show")[2:] if line.startswith("| ")]

    def test_agents_order_breaks_ties_only_and_never_touches_the_task_pins(self):
        self._run("set", "work/a1", "--worker", "Alpha", "--prio", "1")
        self._run("set", "work/b1", "--worker", "Beta", "--prio", "1")
        self._run("set", "work/b2", "--worker", "Beta", "--prio", "2")
        self._run("set", "work/c1", "--worker", "Gamma")
        self._run("set", "work/d1", "--worker", "Delta")
        self._claim("security/gui", "s9", "Epsilon", "line")
        self.assertEqual(
            self._out("agents"),
            ["1\tAlpha\t1\twork/a1", "2\tBeta\t2\twork/b1, work/b2", "3\tGamma\t1\twork/c1", "4\tDelta\t1\twork/d1", "5\tEpsilon\t0\t-"],
        )
        self.assertEqual(self._ranked(), ["work/a1", "work/b1", "work/b2", "work/c1", "work/d1"])
        register = (self.root / "rasci.json").read_text(encoding="utf-8")
        self.assertEqual(
            self._out("agents", "order", "Gamma, Beta,Epsilon"),
            ["agent order: Gamma, Beta, Epsilon; the others follow by register rank; task --prio pins unchanged"],
        )
        self.assertEqual((self.root / "rasci.json").read_text(encoding="utf-8"), register)
        stored = json.loads((self.root / "agent-order.json").read_text(encoding="utf-8"))
        self.assertEqual(stored["agents"], ["Gamma", "Beta", "Epsilon"])
        self.assertEqual(
            self._out("agents"),
            ["1\tGamma\t1\twork/c1", "2\tBeta\t2\twork/b1, work/b2", "3\tEpsilon\t0\t-", "4\tAlpha\t1\twork/a1", "5\tDelta\t1\twork/d1"],
        )
        self.assertEqual(self._ranked(), ["work/b1", "work/a1", "work/b2", "work/c1", "work/d1"])
        self.assertIn("prio: 1", self._out("get", "work/a1"))
        for args in (("agents", "order", "Gamma,Nobody"), ("agents", "order", "Beta,Beta"), ("agents", "order", " , "), ("agents", "sort")):
            with self.subTest(args=args):
                self.assertEqual(self._run(*args).returncode, 2)
        self.assertEqual(json.loads((self.root / "agent-order.json").read_text(encoding="utf-8")), stored)

    def test_the_old_order_command_stores_the_agent_order_and_says_so(self):
        self._run("set", "work/a1", "--worker", "Alpha", "--prio", "3")
        self._run("set", "work/b1", "--worker", "Beta")
        done = self._run("order", "Beta,Alpha")
        self.assertEqual(done.returncode, 0, done.stderr)
        self.assertIn("rasci: `order` is now `agents order`", done.stderr)
        self.assertIn("no longer writes --prio pins", done.stderr)
        self.assertEqual(json.loads((self.root / "agent-order.json").read_text(encoding="utf-8"))["agents"], ["Beta", "Alpha"])
        self.assertIn("prio: 3", self._out("get", "work/a1"))
        self.assertNotIn("prio: 1", self._out("get", "work/b1"))

    def test_a_broken_agent_order_file_ranks_like_none(self):
        self._run("set", "work/a1", "--worker", "Alpha")
        self._run("set", "work/b1", "--worker", "Beta")
        for text in ("{not json", json.dumps({"agents": "Beta"}), json.dumps({"agents": [1, "Beta"]}), json.dumps([1])):
            with self.subTest(text=text):
                (self.root / "agent-order.json").write_text(text, encoding="utf-8")
                self.assertEqual(self._ranked(), ["work/a1", "work/b1"])

    def test_the_agent_order_never_reads_or_writes_through_a_link_fifo_or_deep_nesting(self):
        self._run("set", "work/a1", "--worker", "Alpha")
        self._run("set", "work/b1", "--worker", "Beta")
        outside = Path(self.tmp.name) / "outside.json"
        outside.write_text(json.dumps({"agents": ["Beta"]}), encoding="utf-8")
        order = self.root / "agent-order.json"
        order.symlink_to(outside)
        self.assertEqual(self._ranked(), ["work/a1", "work/b1"])
        order.unlink()
        order.write_text("[" * 100000, encoding="utf-8")
        self.assertEqual(self._ranked(), ["work/a1", "work/b1"])
        order.unlink()
        os.mkfifo(order, 0o600)
        self.assertEqual(self._run("agents", timeout=20).stdout.splitlines(), ["1\tAlpha\t1\twork/a1", "2\tBeta\t1\twork/b1"])
        order.unlink()
        lock = self.root / "rasci.lock"
        lock.unlink(missing_ok=True)
        planted = Path(self.tmp.name) / "planted.lock"
        lock.symlink_to(planted)
        refused = self._run("agents", "order", "Beta,Alpha")
        self.assertEqual(refused.returncode, 2, refused.stderr)
        self.assertIn("agents order", refused.stderr)
        self.assertFalse(planted.exists() or order.exists())

    def test_a_symlinked_register_is_refused_and_never_copied_into_the_council(self):
        outside = Path(self.tmp.name) / "outside.json"
        outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8")
        self.root.mkdir()
        (self.root / "rasci.json").symlink_to(outside)
        for args in (("show", "--all"), ("set", "work/a", "--title", "A")):
            with self.subTest(args=args):
                result = self._run(*args, timeout=60)
                self.assertEqual(result.returncode, 2, result.stderr)
                self.assertNotIn("outside", result.stdout)
        self.assertTrue((self.root / "rasci.json").is_symlink())
        copies = [p.name for p in self.root.iterdir() if p.is_file() and not p.is_symlink()
                  and "marker-outside" in p.read_text(encoding="utf-8")]
        self.assertEqual(copies, [])

    def test_a_hard_linked_register_is_refused(self):
        outside = Path(self.tmp.name) / "outside.json"
        outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8")
        self.root.mkdir()
        os.link(outside, self.root / "rasci.json")
        result = self._run("show", "--all", timeout=60)
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertNotIn("outside", result.stdout)

    def test_a_symlinked_lock_is_refused_and_creates_nothing_at_its_target(self):
        aimed = Path(self.tmp.name) / "elsewhere" / "aimed"
        aimed.parent.mkdir()
        self.root.mkdir()
        (self.root / "rasci.lock").symlink_to(aimed)
        result = self._run("set", "work/a", "--title", "A", timeout=60)
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertFalse(aimed.exists())
        self.assertFalse((self.root / "rasci.json").exists())

    def test_a_fifo_register_fails_instead_of_hanging(self):
        self.root.mkdir()
        os.mkfifo(self.root / "rasci.json")
        for args in (("show",), ("set", "work/a")):
            with self.subTest(args=args):
                self.assertEqual(self._run(*args, timeout=20).returncode, 2)

    def test_a_register_that_is_no_object_of_tasks_exits_2(self):
        self.root.mkdir()
        for text in ("[1, 2]", json.dumps({"work/a": 3}), "[" * 100000):
            with self.subTest(text=text[:12]):
                (self.root / "rasci.json").write_text(text, encoding="utf-8")
                result = self._run("show", "--all", timeout=60)
                self.assertEqual(result.returncode, 2, result.stderr)
                self.assertIn("rasci:", result.stderr)


class TestOrgSkillsUseTheRegister(unittest.TestCase):
    def test_every_org_skill_names_the_register(self):
        for skill in sorted((REPO_ROOT / "skills").glob("kvb-org-*/SKILL.md")):
            if skill.parent.name == "kvb-org-rasci":
                continue
            with self.subTest(skill=skill.parent.name):
                self.assertIn("`kvb-org-rasci`", skill.read_text(encoding="utf-8"))


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