Kernel: spent attempts hold the item even while a refused Run is queued

docs/decisions/0003-one-controller-admits-every-run.md

@@ -28,6 +28,7 @@ single decider has no race to lose and one place for every brake.
 | An approval is a Run with no Engine. One that does not land is bounded by its attempts, and a writer's action renews them | `tests/test_decide.py::test_an_approval_that_does_not_land_is_bounded`, `tests/test_decide.py::test_an_approval_that_landed_is_enqueued_whatever_failed_before`, `tests/test_simulator.py::test_an_approval_that_never_lands_is_bounded` |
 | An approval that lands is an act, and does not count toward the Run ceiling, which is the cost ceiling | `tests/test_decide.py::test_an_approval_run_does_not_count_toward_the_run_ceiling` |
 | An approval that does not land, whatever its fault, counts toward the Run ceiling, so one failing for the environment is not issued for ever | `tests/test_decide.py::test_an_approval_that_keeps_failing_for_the_environment_is_bounded`, `tests/test_simulator.py::test_an_approval_that_keeps_failing_for_the_environment_is_bounded` |
+| A count the completed Runs have spent holds the item, whatever Run waits and whatever would defer; a superseded Run is no attempt, and an act, the act ceiling's among them, waits for the Run | `tests/test_decide.py::test_spent_attempts_hold_the_item_while_a_run_waits`, `tests/test_decide.py::test_spent_rounds_hold_the_item_while_a_run_waits`, `tests/test_decide.py::test_a_spent_count_holds_whatever_would_defer_while_a_run_waits`, `tests/test_decide.py::test_a_run_in_flight_is_held_by_the_completed_runs_alone`, `tests/test_decide.py::test_an_act_waits_for_a_run_in_flight`, `tests/test_simulator.py::test_invariants_hold_under_faults` |
 | A failed model Run is not an approval attempt | `tests/test_decide.py::test_a_failed_model_run_is_not_an_approval_attempt` |
 | Acts since a person with write access acted are held to `max_acts_per_item`; an unreadable count makes no act | `tests/test_decide.py::test_acts_are_bounded`, `tests/test_decide.py::test_an_unreadable_count_of_acts_makes_no_act`, `examples/invalid/facts.no-count-of-acts.json`, `tests/test_simulator.py::test_a_queue_that_ejects_every_time_is_bounded` |
 | The act ceiling stops acts that move work on, and nothing else: a Run is bounded by the Run ceiling, and a Hold is never refused | `tests/test_decide.py::test_the_act_ceiling_stops_acts_and_nothing_else` |

kernel/brakes.py

@@ -48,11 +48,9 @@ def on_act(_decision, facts: dict, state: str, evidence: list, context: dict, ac
     if stopped:
         return stopped
     if action == "issue_approval":
-        failed = _attempts(since, t.ISSUE_APPROVAL)
-        if len(failed) >= limits["max_attempts"]:
-            return _decision(facts, t.HELD, "attempts_exhausted", [r["run"] for r in failed], t.HOLD)
-        if len(_spent(since)) >= limits["max_runs_per_item"]:
-            return _decision(facts, t.HELD, "run_ceiling", [], t.HOLD)
+        held = approval_ceilings(_decision, facts, since, limits)
+        if held:
+            return held
     acts = facts.get("acts_since_writer")
     if type(acts) is not int or acts < 0:
         return _decision(facts, state, "acts_unreadable", evidence, t.DEFER)
@@ -71,6 +69,22 @@ def on_run(_decision, facts: dict, state: str, evidence: list, context: dict, in
     pause = paused(_decision, facts, state, evidence, context)
     if pause:
         return pause
+    return run_ceilings(_decision, facts, evidence, intent, recent, limits)
+
+
+def approval_ceilings(_decision, facts: dict, since: list, limits: dict):
+    """The Hold an approval meets at its attempts or the Run ceiling, or None. No deferral is read here."""
+    failed = _attempts(since, t.ISSUE_APPROVAL)
+    if len(failed) >= limits["max_attempts"]:
+        return _decision(facts, t.HELD, "attempts_exhausted", [r["run"] for r in failed], t.HOLD)
+    if len(_spent(since)) >= limits["max_runs_per_item"]:
+        return _decision(facts, t.HELD, "run_ceiling", [], t.HOLD)
+    return None
+
+
+def run_ceilings(_decision, facts: dict, evidence: list, intent: str, recent: list, limits: dict):
+    """The Hold a Run of this Intent meets at its attempts, its rounds or the Run ceiling, or None. No deferral is read
+    here: decide.py asks it alone while a Run is in flight, when a spent count holds the item whatever would defer."""
     attempts = _attempts(recent, intent)
     if len(attempts) >= limits["max_attempts"]:
         return _decision(facts, t.HELD, "attempts_exhausted", [r["run"] for r in attempts], t.HOLD)

kernel/decide.py

@@ -67,9 +67,14 @@ def _wanted_run(state: str, reason: str, config: dict):
     return None
 
 
+def _action(reason: str) -> str:
+    """The act that moves a converged item on."""
+    return "enqueue" if reason == "approved" else "issue_approval"
+
+
 def _act(facts: dict, state: str, reason: str, evidence: list, mine: list, context: dict, config: dict) -> dict:
     """An act that moves work on, once the brakes let it through. An act that stops work is not decided here."""
-    action = "enqueue" if reason == "approved" else "issue_approval"
+    action = _action(reason)
     since = since_writer(mine, facts)
     brake = brakes.on_act(_decision, facts, state, evidence, context, action, since, config["limits"])
     if brake:
@@ -109,7 +114,20 @@ def _decide(facts: dict, runs: list, context: dict, config: dict, now: str) -> d
         return _decision(facts, state, reason, evidence, t.HOLD)
 
     mine = runs_on(runs, facts)
+    wanted = _wanted_run(state, reason, config)
     if in_progress(mine):
+        # No Run in flight lowers a count: attempts, rounds and the Run ceiling count completed Runs, and only a
+        # writer's action renews them. So a count the completed Runs have spent holds the item, whatever Run waits and
+        # whatever would defer, a Pause among them: the ceilings are asked alone (decision 0003). The act ceiling is
+        # not, since it counts acts and stops nothing but an act, and none is decided while a Run waits.
+        done = [r for r in since_writer(mine, facts) if r["status"] == "completed"]
+        held = None
+        if state == t.CONVERGED and _action(reason) == "issue_approval":
+            held = brakes.approval_ceilings(_decision, facts, done, config["limits"])
+        elif wanted is not None:
+            held = brakes.run_ceilings(_decision, facts, evidence, wanted[1], done, config["limits"])
+        if held:
+            return held
         return _decision(facts, state, reason, evidence, t.NOTHING, working=True)
 
     if state == t.CONVERGED:
@@ -119,7 +137,6 @@ def _decide(facts: dict, runs: list, context: dict, config: dict, now: str) -> d
     # A required check that gave no verdict is unreadable: admit nothing, and wait for a person (I-13).
     if reason == "check_unreadable":
         return _decision(facts, state, reason, evidence, t.DEFER)
-    wanted = _wanted_run(state, reason, config)
     if wanted is None:
         return _decision(facts, state, reason, evidence, t.NOTHING)
     role, intent = wanted

sim/AGENTS.md

@@ -30,7 +30,7 @@ moment, results on earlier content stay behind, a second result can disagree wit
 the contract short or with a result on this content for a criterion it does not name. A person approves the contract, or on some seeds has not yet,
 and can withdraw it; a push can edit or delete it; another App or workflow can give a result, and a check one for a criterion only a person passes,
 which on an agent's pull request a person's approval of the content confirms (`tests/test_simulator.py::test_the_seeds_serve_items_with_an_acceptance_contract_and_without`).
-On one seed in six a review's Result now and then leaves a criterion of its rubric uncovered: its Run fails by a fault of the item and opens no thread,
+On one seed in three a review's Result now and then leaves a criterion of its rubric uncovered: its Run fails by a fault of the item and opens no thread,
 and the review is admitted again until its attempts hold the item (`tests/test_simulator.py::test_the_seeds_serve_a_review_that_left_a_criterion_uncovered`).
 A rule about anything else is not exercised here, so its boundary cases are all that hold it. An invariant is a call to `check()` with its name. When a seed finds a fault, the fix goes in the
 Kernel with a boundary case in `tests/test_decide.py`.

sim/world.py

@@ -82,13 +82,13 @@
 ON_THE_LIST = {"role": "tests/test_decide.py", "dependency": ".github/workflows/verify.yml",
                "maintainer": "config/defaults.json"}
 OFF_THE_LIST = {"role": "docs/lessons.md", "dependency": "docs/requirements.txt", "maintainer": "docs/glossary.md"}
-# On one seed in six, under faults, a review's Result leaves a criterion of its rubric uncovered half the time, which
+# On one seed in three, under faults, a review's Result leaves a criterion of its rubric uncovered half the time, which
 # `apply` refuses before any Effect: the Run fails by a fault of the item (decision 0029). The Factory's own behaviour,
-# not GitHub's, so it carries no marker. By the seed alone, so every other world stays as it was. The spread is narrowed
-# for "Kernel: spent attempts hold the item even while a refused Run is queued" (issue 158): on more seeds, a Run list
-# missing its newest Run admits a review past its attempts, and the State leaves the Hold while that Run waits. The fix
-# for that issue widens it again.
-UNCOVERED_SEEDS, UNCOVERED = 6, 0.5
+# not GitHub's, so it carries no marker. By the seed alone, so every other world stays as it was. On every seed, all 300
+# hold, but the first 40, which a local run and the mutation check use, no longer reach every path the other tests ask
+# for: test_invariants_hold_under_faults sees no approval stopped, and
+# test_the_seeds_serve_items_with_an_acceptance_contract_and_without no acceptance_evidence_missing.
+UNCOVERED_SEEDS, UNCOVERED = 3, 0.5
 
 
 class Broken(AssertionError):
@@ -612,7 +612,12 @@ def paused(self, facts: dict, runs: list, d: dict):
         self.check(d["outcome"] not in (t.ADMIT, t.ACT), "a Pause let through %s %s" % (d["outcome"], d["reason"]))
         free = decide(facts, runs, dict(self.context(), paused=False), CONFIG, self.now())
         why = "paused" if self.pause_reading else "pause_unreadable"
-        if free["outcome"] in (t.ADMIT, t.ACT) or (free["outcome"] == t.HOLD and free["reason"] in CEILINGS):
+        ceiling = free["outcome"] == t.HOLD and free["reason"] in CEILINGS
+        # A Run of the item in flight, as the Kernel reads it: admitted and not completed. While one is, a count already
+        # spent holds the item whatever would defer, a Pause among them, so the Hold shows through (issue 158).
+        if ceiling and any(r["item"] == facts["item"] and r["status"] != "completed" for r in runs):
+            self.check(d == free, "a spent count is hidden by a Pause while a Run is in flight")
+        elif free["outcome"] in (t.ADMIT, t.ACT) or ceiling:
             self.stopped.append(free.get("action") or free["outcome"])
             self.check((d["outcome"], d["reason"]) == (t.DEFER, why),
                        "a Pause did not defer %s %s" % (free["outcome"], free["reason"]))

tests/test_decide.py

@@ -2038,6 +2038,106 @@ def test_with_no_writers_action_every_run_counts(self):  # I-8, I-28
         d = go(none, REVIEWED + approvals(LIMITS["max_attempts"] - 1))
         self.assertEqual((d["outcome"], d["action"]), ("act", "issue_approval"))
 
+    def test_spent_attempts_hold_the_item_while_a_run_waits(self):  # I-8, I-12, I-23
+        # The simulator's seed 297: a review superseded, three that failed by a fault of the item, and a fourth admitted
+        # on a list read one Run short. No Run in flight lowers a count, so the item is held while it waits, and a
+        # superseded Run is no attempt.
+        r0 = run(conclusion="failure", outcome="superseded", at=at(10), name="r0")
+        failed = [run(conclusion="failure", outcome="item", at=at(11 + i), name="r%d" % (i + 1)) for i in range(3)]
+        for status in ("queued", "in_progress"):
+            r4 = dict(run(status="in_progress", at=at(14), name="r4"), status=status)
+            d = go(facts(), [r0] + failed + [r4])
+            self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"], d["evidence"]),
+                             ("held", "attempts_exhausted", "hold", "needs_you", ["r1", "r2", "r3"]), status)
+            # The other side: without r3 the attempts are not spent, and r0 is none of them. The Run waits.
+            d = go(facts(), [r0] + failed[:2] + [r4])
+            self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                             ("verifying", "awaiting_review", "nothing", "working"), status)
+        # When the lane supersedes r4, as `apply` does on a held item, the evidence is still r1 to r3, and it stays held.
+        d = go(facts(), [r0] + failed + [run(conclusion="failure", outcome="superseded", at=at(14), name="r4")])
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["evidence"]),
+                         ("held", "attempts_exhausted", "hold", ["r1", "r2", "r3"]))
+        # A converged item's approval attempts hold it the same way, whatever Run waits.
+        waiting = [dict(run(intent="issue-approval", status="in_progress", at=at(40), name="a9"), status="queued")]
+        failed = approvals(LIMITS["max_attempts"])
+        d = go(facts(), REVIEWED + failed + waiting)
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["evidence"]),
+                         ("held", "attempts_exhausted", "hold", [r["run"] for r in failed]))
+        d = go(facts(), REVIEWED + failed[1:] + waiting)
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                         ("converged", "approval_due", "nothing", "working"))
+
+    def test_a_spent_count_holds_whatever_would_defer_while_a_run_waits(self):  # I-8, I-12
+        # A Pause, one that cannot be read, or an unreadable count of acts does not hide a count already spent while a
+        # Run is in flight: the ceilings are asked alone.
+        r0 = run(conclusion="failure", outcome="superseded", at=at(10), name="r0")
+        failed = [run(conclusion="failure", outcome="item", at=at(11 + i), name="r%d" % (i + 1)) for i in range(3)]
+        approvals_failed = approvals(LIMITS["max_attempts"])
+        waiting = [dict(run(intent="issue-approval", status="in_progress", at=at(40), name="a9"), status="queued")]
+        for status in ("queued", "in_progress"):
+            r4 = dict(run(status="in_progress", at=at(14), name="r4"), status=status)
+            for paused in (True, None):
+                d = go(facts(), [r0] + failed + [r4], ctx=context(paused=paused))
+                self.assertEqual((d["state"], d["reason"], d["outcome"], d["evidence"]),
+                                 ("held", "attempts_exhausted", "hold", ["r1", "r2", "r3"]), (status, paused))
+                d = go(facts(), REVIEWED + approvals_failed + waiting, ctx=context(paused=paused))
+                self.assertEqual((d["state"], d["reason"], d["outcome"]), ("held", "attempts_exhausted", "hold"), paused)
+                # The other side: with the count not spent, the Run in flight keeps the item working, as it did.
+                d = go(facts(), [r0] + failed[:2] + [r4], ctx=context(paused=paused))
+                self.assertEqual((d["state"], d["outcome"], d["attention"]), ("verifying", "nothing", "working"))
+        d = go(facts(acts_since_writer=None), REVIEWED + approvals_failed + waiting)
+        self.assertEqual((d["state"], d["reason"], d["outcome"]), ("held", "attempts_exhausted", "hold"))
+        d = go(facts(acts_since_writer=None), REVIEWED + approvals_failed[1:] + waiting)
+        self.assertEqual((d["state"], d["outcome"], d["attention"]), ("converged", "nothing", "working"))
+
+    def test_an_act_waits_for_a_run_in_flight(self):  # I-12
+        # The act ceiling counts acts, not Runs, and stops only an act: none is decided while a Run waits.
+        ceiling = facts(acts_since_writer=LIMITS["max_acts_per_item"])
+        waiting = dict(run(intent="adjudicate-findings", status="in_progress", at=at(40), name="o9"), status="queued")
+        d = go(ceiling, REVIEWED + [waiting])
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                         ("converged", "approval_due", "nothing", "working"))
+        d = go(ceiling, REVIEWED)
+        self.assertEqual((d["state"], d["reason"], d["outcome"]), ("held", "act_ceiling", "hold"))
+        # An approval that landed is enqueued whatever failed before, so failed approvals do not hold it either.
+        landed = facts(reviews=APPROVED)
+        d = go(landed, REVIEWED + approvals(LIMITS["max_attempts"]) + [waiting])
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                         ("converged", "approved", "nothing", "working"))
+
+    def test_spent_rounds_hold_the_item_while_a_run_waits(self):  # I-14
+        # An item whose owner's rulings are at `max_rounds` is held while another Run is queued or in progress, as with
+        # none in flight. One round short, the Run in flight keeps it working.
+        rounds = LIMITS["max_rounds"]
+        laps = [r for i in range(rounds) for r in (run(intent="adjudicate-findings", at=at(20 + 2 * i), name="o%d" % i),
+                                                   run(role="dev-lead", intent="fix-findings", at=at(21 + 2 * i),
+                                                       name="a%d" % i))]
+        short = [r for r in laps if r["run"] not in ("o%d" % (rounds - 1), "a%d" % (rounds - 1))]
+        for status in ("queued", "in_progress"):
+            waiting = dict(run(intent="adjudicate-findings", status="in_progress", at=at(40), name="o9"), status=status)
+            d = go(facts(threads=[thread()]), REVIEWED + laps + [waiting])
+            self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                             ("held", "max_rounds", "hold", "needs_you"), status)
+            d = go(facts(threads=[thread()]), REVIEWED + short + [waiting])
+            self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                             ("adjudicating", "proposals_await_owner", "nothing", "working"), status)
+        # The other side of the Run in flight: with none, the same rounds hold it and one short admits a ruling.
+        self.assertEqual(go(facts(threads=[thread()]), REVIEWED + laps)["reason"], "max_rounds")
+        self.assertEqual(go(facts(threads=[thread()]), REVIEWED + short)["admit"]["intent"], "adjudicate-findings")
+
+    def test_a_run_in_flight_is_held_by_the_completed_runs_alone(self):  # I-8
+        # The Run ceiling counts the Runs that completed: the tenth Run, admitted with nine completed, is not held at
+        # the ceiling it was admitted under. Ten completed hold the item while another waits.
+        ceiling = LIMITS["max_runs_per_item"]
+        spent = [run(conclusion="failure", outcome="environment", at=at(10 + i), name="e%d" % i) for i in range(ceiling)]
+        waiting = dict(run(status="in_progress", at=at(40), name="w"), status="queued")
+        d = go(facts(), spent[1:] + [waiting])
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                         ("verifying", "awaiting_review", "nothing", "working"))
+        d = go(facts(), spent + [waiting])
+        self.assertEqual((d["state"], d["reason"], d["outcome"], d["attention"]),
+                         ("held", "run_ceiling", "hold", "needs_you"))
+
     def test_an_approval_that_does_not_land_is_bounded(self):  # I-8, I-28
         failed = approvals(LIMITS["max_attempts"])
         d = go(facts(), REVIEWED + failed)