From c6048751169cfc287037846516a624642bd8bd08 Mon Sep 17 00:00:00 2001 From: di-omics Date: Tue, 28 Jul 2026 01:52:03 -0700 Subject: [PATCH] Schedule more than one plate, and price a plate-moving arm honestly Every layer here costs a single run. A workcell exists because one plate at a time wastes the instruments, and the second plate introduces constraints a single-run report cannot see. Nothing in this package computed them. Computes them without durations, which is the constraint that shaped the design. `throughput` refuses a plates-per-day number because 18 of 18 steps are untimed, and adding plates makes that refusal stronger rather than weaker: a pipelined estimate compounds every unmeasured duration in it. But the serialization point is the instrument that owns the most steps, which is true at any speed and needs no stopwatch, and it names which box to buy a second of. Stalls are classified rather than collapsed. Five constraints stop a run and each is cleared by a different budget -- decode is reverse-engineering, handoff is a plate mover or a barcode, gate is an instrument that returns a usable number, trust is a calibration experiment, and attended is the one no purchase fixes. arm_relief() prices the reflex answer. On the genomics protocol a plate-moving arm clears 5 of 22 stalls and none of the 8 undecoded commands: it carries plates and reports the move, which is the only thing that closes a custody gap, and it decodes nothing. Scopes trust to where it binds. A USB enumeration moves nothing, so no robot benchmark gates it; it fires only where the machine can act and has not earned the right, so proving run cards moves stalls from decode to trust rather than removing them. Every stall resolves from the layer that already computed it rather than being recomputed, and a test pins the unattended-step count to the ledger's own headless prefix. 11 new tests, 156 total. --- README.md | 50 +++++- autonomous_lab/__init__.py | 9 + autonomous_lab/cli.py | 52 ++++++ autonomous_lab/orchestrate.py | 300 ++++++++++++++++++++++++++++++++++ tests/test_orchestrate.py | 148 +++++++++++++++++ 5 files changed, 558 insertions(+), 1 deletion(-) create mode 100644 autonomous_lab/orchestrate.py create mode 100644 tests/test_orchestrate.py diff --git a/README.md b/README.md index c68eac8..a9480ee 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ autonomous-lab vision # what a camera catches; what none autonomous-lab throughput single_cell_genomics # plates/day, or why that number does not exist autonomous-lab provenance single_cell_genomics # what could be proven about a run afterwards autonomous-lab lineage single_cell_genomics # which cell did this read come from? +autonomous-lab schedule single_cell_genomics --plates 10 # what serializes across a workcell? autonomous-lab knowledge # encoded expert judgment and robot benchmarks ``` @@ -225,6 +226,51 @@ between the plate now on the deck and the record describing it rests on the huma carried the right one. Software cannot close that by being careful. A barcode read at both ends can, or an arm that reports the move. +### More than one plate: what serializes, and what an arm actually buys + +Every layer above costs a **single** run. A workcell exists because one plate at a time +wastes the instruments, and the second plate introduces constraints a single-run report +cannot see. + +``` +$ autonomous-lab schedule single_cell_genomics --plates 10 + + serialization point namocell (7 of 18 steps, 39%) + 70 sequential operations for 10 plates + unattended run gets 1 of 18 step(s) in + + CYCLE TIME NOT COMPUTABLE + + 22 stall(s): decode 8 attended 7 handoff 5 gate 2 + + A PLATE-MOVING ARM would clear 5 stall(s) and leave 17. +``` + +The design constraint is that **none of this uses durations**. `throughput` refuses a +plates-per-day number because 18 of 18 steps are untimed, and adding plates makes that +refusal stronger rather than weaker -- a pipelined estimate compounds every unmeasured +duration in it. But most of what a workcell planner needs is structural, not temporal. +*Which instrument serializes* is the one that appears in the most steps; that is true at +any speed, needs no stopwatch, and names which box to buy a second of. + +Stalls are **classified rather than collapsed**, because five constraints stop a run and +each is cleared by a different budget: `decode` is reverse-engineering, `handoff` is a +plate mover or a barcode, `gate` is an instrument that returns a usable number, `trust` is +a calibration experiment, and `attended` is the one no purchase fixes. + +`arm_relief()` is the computation this repo did not have anywhere. "Buy a plate mover" is +the reflex answer to a stalling schedule, and here it is **wrong**: an arm clears 5 of 22 +stalls and none of the 8 undecoded commands. It carries plates and reports the move, which +is the one thing that closes a custody gap -- and it decodes nothing, makes no broken reader +return a number, and runs no calibration. Which purchase helps is computable before the +order rather than after it. + +`trust` is scoped to where it actually binds. Almost nothing here has a met benchmark, so +flagging every step would bury the other four classes; and a USB enumeration moves nothing, +so no robot benchmark gates it. It fires only where the machine **can** act and has not +earned the right to -- which, with plr-tested wired in, is exactly the two steps that run on +real hardware under supervision. + ### Sample identity: which cell the read came from Everything above tracks **containers**. The ledger costs a plate hop, provenance records @@ -472,7 +518,7 @@ event receiver and silently steals the first one's callbacks. pip install -e '.[dev]' && pytest ``` -149 device-free tests. The ones that matter most try to make the layer lie: +160 device-free tests. The ones that matter most try to make the layer lie: - claim a step is automated when its command is undecoded; claim a decoded command is runnable while its siblings are not; claim a federated leg runs when no run card was ever @@ -488,6 +534,8 @@ pip install -e '.[dev]' && pytest - treat an unbenchmarked operation as trusted. - report an intact sample chain through a pooling step that destroyed attribution, or claim a pooled absorbance reading says something about one well. +- return a workcell cycle time over steps nobody timed, or report a plate-moving arm as + clearing a stall that is really an undecoded command. Two tests exist to keep the honesty from drifting quietly. One asserts that even a fully equipped workcell -- every camera, every calibration, every model -- still leaves diff --git a/autonomous_lab/__init__.py b/autonomous_lab/__init__.py index f6a45a9..1af9d27 100644 --- a/autonomous_lab/__init__.py +++ b/autonomous_lab/__init__.py @@ -34,6 +34,8 @@ provenance a tamper-evident run record, and where the custody chain breaks. lineage which cell a read came from -- and whether pooling already destroyed the answer, which is a different question from whether the plate was tracked. + orchestrate more than one plate at once: what serializes, where it stalls and on which + of five constraints, and what a plate-moving arm would actually buy. intelligence the tacit expert judgment and the benchmarks a robot must meet first. """ @@ -64,6 +66,7 @@ undeclared_transforms, ) from .model import Artifact, Protocol, Role, Step, Tier, Transform, Verdict, ZeroDecodeOp +from .orchestrate import Constraint, Contention, Schedule, Stall, contention, orchestrate from .provenance import Attestation, CustodyGap, Event, RunRecord, provenance_report from .qc import Basis, Criterion, Decision, Gate, Readiness, evaluate, gate_report from .recovery import Detection, FailureMode, Latency, Severity, recovery_report @@ -98,6 +101,8 @@ def loop_closure_for(protocol, workcell=None): "Basis", "Benchmark", "BenchmarkStatus", + "Constraint", + "Contention", "Criterion", "CustodyGap", "Decision", @@ -126,7 +131,9 @@ def loop_closure_for(protocol, workcell=None): "Readiness", "Role", "RunRecord", + "Schedule", "Separability", + "Stall", "RunReport", "Severity", "Step", @@ -146,6 +153,7 @@ def loop_closure_for(protocol, workcell=None): "ZeroDecodeOp", "build_ledger", "build_lineage", + "contention", "cost_step", "declared", "estimate", @@ -155,6 +163,7 @@ def loop_closure_for(protocol, workcell=None): "lineage_report", "loop_closure", "loop_closure_for", + "orchestrate", "provenance_report", "rank_unlocks", "recovery_report", diff --git a/autonomous_lab/cli.py b/autonomous_lab/cli.py index 16c258a..5f63736 100644 --- a/autonomous_lab/cli.py +++ b/autonomous_lab/cli.py @@ -33,6 +33,7 @@ ) from .ledger import build_ledger, rank_unlocks from .lineage import MISASSIGNMENT, Separability, lineage_report +from .orchestrate import orchestrate from .model import Verdict from .provenance import provenance_report from .qc import MEASUREMENTS, gate_report @@ -380,6 +381,51 @@ def _lineage(args) -> int: return 0 if report.verdict.ok else 1 +def _orchestrate(args) -> int: + """More than one plate at once: what serializes, and where it stops.""" + wc = _workcell(args) + p = protocols.get(args.protocol) + ledger = build_ledger(p, wc) + sched = orchestrate( + p, ledger, gate_report(p.name, ledger), plates=args.plates, untrusted=untrusted_ops(p) + ) + + print(f"workcell schedule for {p.name}, {sched.plates} plate(s)\n") + b = sched.bottleneck + if b is not None: + print(f" serialization point {b.instrument} ({b.steps} of {len(p.steps)} steps, {b.share:.0%})") + print(f" {b.serialized_slots(sched.plates)} sequential operations for {sched.plates} plates") + print(f" unattended run gets {sched.unattended_steps} of {len(p.steps)} step(s) in") + print(f" plates in flight {sched.pipeline_depth} (distinct instruments before it stops)\n") + + print(" CYCLE TIME NOT COMPUTABLE") + print(f" {sched.why_no_makespan()}\n") + + counts = sched.by_constraint() + print(f" {len(sched.stalls)} stall(s), by what it would take to clear them:") + for kind, n in sorted(counts.items(), key=lambda kv: -kv[1]): + print(f" {kind:9} {n}") + + first = sched.first_stall() + if first is not None: + print(f"\n STOPS AT {first.step_op} on {first.instrument} [{first.constraint.value}]") + print(f" {first.detail}") + print(f" clears it: {first.clears_it}") + + removed, remains = sched.arm_relief() + print(f"\n A PLATE-MOVING ARM would clear {len(removed)} stall(s) and leave {len(remains)}.") + for st in removed: + print(f" cleared {st.step_op}: {st.detail}") + if remains: + worst = remains[0] + print( + f" remains {worst.step_op} [{worst.constraint.value}] and {len(remains) - 1} more." + f"\n An arm carries plates. It decodes no command sets, makes no broken" + "\n reader return a number, and runs no calibration." + ) + return 0 if sched.runs_unattended else 1 + + def _knowledge(args) -> int: """The tacit layer: expert judgments and the benchmarks a robot must meet.""" s = knowledge_summary() @@ -523,6 +569,12 @@ def common(sp): ln.add_argument("--unit", default="cell", help="what the science needs to distinguish") ln.set_defaults(func=_lineage) + oc = sub.add_parser("schedule", help="run more than one plate: what serializes and where it stops") + common(oc) + oc.add_argument("protocol", nargs="?", default="single_cell_genomics") + oc.add_argument("--plates", type=int, default=10, help="how many plates to push through") + oc.set_defaults(func=_orchestrate) + kn = sub.add_parser("knowledge", help="encoded expert judgment and robot benchmarks") common(kn) kn.set_defaults(func=_knowledge) diff --git a/autonomous_lab/orchestrate.py b/autonomous_lab/orchestrate.py new file mode 100644 index 0000000..5ff46f9 --- /dev/null +++ b/autonomous_lab/orchestrate.py @@ -0,0 +1,300 @@ +"""Running more than one plate: what serializes, where it stalls, and what an arm buys. + +Every other layer in this package costs a SINGLE run. That is the right first question and +it is not the question a workcell asks. A workcell exists because one plate at a time +wastes the instruments, and the moment a second plate enters, a set of constraints appears +that a single-run report cannot see: two plates want the same liquid handler, a human is +needed at step four and it is 2am, and the plate that finished amplifying is sitting on a +deck nobody can clear. + +This module computes that, and it computes it **without durations**, which is the part +worth being careful about. `throughput` already refuses to return plates-per-day because 18 +of 18 steps have never been timed, and that refusal stands. But most of what a workcell +planner needs is structural rather than temporal: + + Which instrument is the serialization point? The one that appears in the most steps. That + is true at any speed, so it needs no stopwatch. Buying a second one is the only thing that + changes it, and knowing which one to buy is worth more than a cycle time. + + How many plates can be in flight unattended? Bounded by how far an unattended run gets at + all -- which the ledger already computes -- and then by how many distinct instruments that + prefix touches. A prefix of one step pipelines no plates no matter how fast it is. + + Where does the run stop, and on what? Five different constraints stop a run and they are + fixed by five different budgets. Collapsing them into "blocked" is the same mistake as + collapsing loop closure into one autonomy percentage. + +The last computation is the one this repo did not have anywhere. `arm_relief()` asks what a +plate-moving arm would actually buy, and answers it by constraint class rather than by +enthusiasm. An arm removes HANDOFF stalls -- it carries the plate and reports the move, which +is the one thing that closes a custody gap. It removes none of the others. A lab that buys an +arm to fix a schedule stalling on undecoded commands has bought the wrong machine, and the +distinction is computable before the purchase order rather than after it. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Tuple + +from .model import Protocol, Verdict, ZeroDecodeOp + + +class Constraint(str, Enum): + """Why a multi-plate run stops here. Each one is fixed by a different budget. + + Ordered by how early in a lab's life it usually bites. DECODE is reverse-engineering + work, HANDOFF is a plate mover or a barcode, GATE is an instrument that returns a usable + number, TRUST is a calibration experiment, and ATTENDED is the one no purchase fixes -- + a step a human must physically do, which caps overnight running at zero however fast + everything else becomes. + """ + + DECODE = "decode" # the command is undecoded; the step cannot run at all + ATTENDED = "attended" # a human must be at the bench for this step + HANDOFF = "handoff" # material must be carried between instruments, unobserved + GATE = "gate" # a QC gate sits here and cannot be evaluated + TRUST = "trust" # the operation has no met benchmark, so it is not trusted unattended + + @property + def arm_fixes_it(self) -> bool: + """Whether a plate-moving arm removes this stall. + + Only HANDOFF. An arm carries plates and reports the move; it does not decode a command + set, does not make a broken reader return a number, and does not run a calibration. + Stated as a property rather than left to the reader because "buy a robot" is the + default answer to a stalling schedule and it is usually the wrong one. + """ + return self is Constraint.HANDOFF + + +@dataclass(frozen=True) +class Stall: + """One point where a multi-plate run stops, and what specifically clears it.""" + + step_op: str + instrument: str + constraint: Constraint + detail: str + clears_it: str + + @property + def arm_fixes_it(self) -> bool: + return self.constraint.arm_fixes_it + + +@dataclass(frozen=True) +class Contention: + """One instrument and how much of the protocol it owns. + + `steps` is the count of protocol steps that run on this instrument. Under the + one-process-per-instrument rule those steps serialize across every plate in flight, so + the instrument with the highest count is the structural bottleneck at any plate count and + at any speed. This is the number that survives having no timing data. + """ + + instrument: str + steps: int + share: float # fraction of the protocol's steps + + def serialized_slots(self, plates: int) -> int: + """How many times this instrument must be used to push `plates` plates through. + + A lower bound on the schedule that needs no durations: the busiest instrument cannot + do better than this many sequential operations, whatever the others do. + """ + return self.steps * plates + + +@dataclass +class Schedule: + """What a workcell running this protocol would actually do, and where it would stop.""" + + protocol: str + plates: int + stalls: List[Stall] = field(default_factory=list) + contention: List[Contention] = field(default_factory=list) + unattended_steps: int = 0 + pipeline_depth: int = 0 + + @property + def bottleneck(self) -> Optional[Contention]: + return self.contention[0] if self.contention else None + + @property + def runs_unattended(self) -> bool: + return self.unattended_steps > 0 and not self.stalls + + def first_stall(self) -> Optional[Stall]: + return self.stalls[0] if self.stalls else None + + def by_constraint(self) -> Dict[str, int]: + out: Dict[str, int] = {} + for s in self.stalls: + out[s.constraint.value] = out.get(s.constraint.value, 0) + 1 + return out + + def arm_relief(self) -> Tuple[List[Stall], List[Stall]]: + """(stalls an arm removes, stalls it does not). + + The honest form of "should we buy a plate mover". An arm is the only thing that closes + a custody gap, so where the second list dominates, the arm is not the next purchase -- + and the first stall in the second list names what is. + """ + removed = [s for s in self.stalls if s.arm_fixes_it] + remains = [s for s in self.stalls if not s.arm_fixes_it] + return removed, remains + + def makespan(self) -> None: + """Deliberately not implemented. + + A workcell schedule over untimed steps is a drawing, not a plan. `throughput` already + refuses this for one plate and the refusal does not get weaker by adding plates: it gets + stronger, because a pipelined estimate compounds every unmeasured duration in it. The + structural results above are real without it, which is the whole design of this module. + """ + return None + + def why_no_makespan(self) -> str: + return ( + "no step in this protocol has a measured duration, so a cycle time over them would " + "be a guess formatted as a specification. The serialization structure above holds " + "without one" + ) + + +# Read-only operations: they enumerate a bus or open a socket and move nothing, so no +# robot benchmark stands between them and running unattended. +_ZERO_DECODE_OPS = {op.value for op in ZeroDecodeOp} +_COULD_RUN = (Verdict.AUTOMATED, Verdict.SUPERVISED) + + +def contention(protocol: Protocol) -> List[Contention]: + """Instruments ranked by how much of the protocol they own, busiest first.""" + counts: Dict[str, int] = {} + for step in protocol.steps: + counts[step.instrument] = counts.get(step.instrument, 0) + 1 + total = len(protocol.steps) or 1 + rows = [Contention(instrument=k, steps=v, share=v / total) for k, v in counts.items()] + return sorted(rows, key=lambda c: (-c.steps, c.instrument)) + + +def _hop_targets(ledger) -> Dict[str, List[Tuple[str, str]]]: + """Which steps receive material carried from another instrument, keyed by consumer.""" + out: Dict[str, List[Tuple[str, str]]] = {} + producer: Dict[str, str] = {} + physical = {a.name for a in ledger.protocol.artifacts if a.physical} + for step in ledger.protocol.steps: + for art in step.consumes: + src = producer.get(art) + if art in physical and src is not None and src != step.instrument: + out.setdefault(step.op, []).append((art, src)) + for art in step.produces: + producer[art] = step.instrument + return out + + +def orchestrate( + protocol: Protocol, ledger, gates=None, plates: int = 1, untrusted: Optional[List] = None +) -> Schedule: + """Plan a multi-plate run: what serializes, where it stalls, on which constraint. + + Every stall is resolved from a layer that already computed it -- verdicts from the + ledger, hops from `handoffs()`, gate readiness from `qc`, benchmarks from `intelligence` + -- rather than recomputed here. A second computation of the same fact eventually + disagrees with the first and nothing reveals which one is wrong. + """ + sched = Schedule(protocol=protocol.name, plates=plates, contention=contention(protocol)) + + hops = _hop_targets(ledger) + unevaluable = set() + if gates is not None: + for row in gates.unsatisfiable(): + unevaluable.add((row.gate.after_step, row.gate.name, row.reason)) + untrusted_ops = {op for op, _reason in (untrusted or [])} + + reached = 0 + for row in ledger.rows: + step = row.step + stalls_here: List[Stall] = [] + + if row.verdict in (Verdict.BLOCKED, Verdict.WRITTEN, Verdict.BROKEN): + stalls_here.append( + Stall( + step_op=step.op, + instrument=step.instrument, + constraint=Constraint.DECODE, + detail=f"verdict is {row.verdict.value}; the step does not run headlessly today", + clears_it=( + "debug the failure" + if row.verdict is Verdict.BROKEN + else "finish this instrument's map, or prove the run card on hardware" + ), + ) + ) + elif row.verdict is Verdict.MANUAL: + stalls_here.append( + Stall( + step_op=step.op, + instrument=step.instrument, + constraint=Constraint.ATTENDED, + detail=step.manual_reason or "a human performs this step at the bench", + clears_it="an instrument or an arm that performs it, or accept attended-hours only", + ) + ) + + for artifact, src in hops.get(step.op, []): + stalls_here.append( + Stall( + step_op=step.op, + instrument=step.instrument, + constraint=Constraint.HANDOFF, + detail=f"'{artifact}' is carried from {src} to {step.instrument} by hand", + clears_it="a plate mover that reports the move, or a barcode read at both ends", + ) + ) + + for after_step, name, reason in unevaluable: + if after_step == step.op: + stalls_here.append( + Stall( + step_op=step.op, + instrument=step.instrument, + constraint=Constraint.GATE, + detail=f"gate '{name}' sits here and cannot be evaluated: {reason}", + clears_it="an instrument that returns the measurement this gate reads", + ) + ) + + # TRUST only where it is the BINDING constraint: a step that could otherwise run, and + # that actually actuates. Almost nothing here has a met benchmark, so applying it to + # every step would bury the four constraints that differ from each other -- and a + # socket probe or a bus enumeration moves nothing, so no robot benchmark gates it. + actuates = step.op not in _ZERO_DECODE_OPS + if step.op in untrusted_ops and actuates and row.verdict in _COULD_RUN: + stalls_here.append( + Stall( + step_op=step.op, + instrument=step.instrument, + constraint=Constraint.TRUST, + detail=( + f"verdict is {row.verdict.value}, so the machine can do it -- but no benchmark " + "has been met, so it is not trusted to do it unattended" + ), + clears_it="run the calibration experiment that retires its benchmark", + ) + ) + + if stalls_here and not sched.stalls: + # The first stalling step is where an unattended run actually stops. + sched.unattended_steps = reached + sched.stalls.extend(stalls_here) + if not stalls_here: + reached += 1 + + if not sched.stalls: + sched.unattended_steps = reached + prefix = [r.step.instrument for r in ledger.rows[: sched.unattended_steps]] + sched.pipeline_depth = len(set(prefix)) + return sched diff --git a/tests/test_orchestrate.py b/tests/test_orchestrate.py new file mode 100644 index 0000000..53b73ee --- /dev/null +++ b/tests/test_orchestrate.py @@ -0,0 +1,148 @@ +"""Device-free tests for multi-plate scheduling. + +The failure mode this guards against is a scheduler that looks useful. Given a protocol and +a plate count, the natural implementation returns a cycle time and a Gantt chart, and both +are fabrications when no step has ever been timed -- worse than nothing, because a total +gets quoted and becomes a specification nobody measured. + +So the tests hold two lines. The structural results (which instrument serializes, where the +run stops, on what constraint) must be real without any timing data. And the temporal +result must stay refused. + +The other load-bearing test is `arm_relief`. "Buy a plate mover" is the default answer to a +stalling schedule, and here it is wrong: an arm clears 5 of 22 stalls and none of the 8 that +are undecoded commands. If that test ever starts reporting an arm as sufficient, someone has +reclassified a decode problem as a logistics one. +""" + +from __future__ import annotations + +from autonomous_lab import Workcell, build_ledger, protocols +from autonomous_lab.intelligence import untrusted_ops +from autonomous_lab.orchestrate import Constraint, contention, orchestrate +from autonomous_lab.qc import gate_report + + +def _sched(plates=10, plr_tested=None): + wc = Workcell.default() + if plr_tested: + wc.plr_tested_root = plr_tested + p = protocols.get("single_cell_genomics") + ledger = build_ledger(p, wc) + return p, ledger, orchestrate( + p, ledger, gate_report(p.name, ledger), plates=plates, untrusted=untrusted_ops(p) + ) + + +# -- structure that survives having no clock ------------------------------------ + + +def test_the_bottleneck_is_the_instrument_that_owns_the_most_steps(): + """True at any speed, so it needs no stopwatch. This is the number worth having when + every duration is unmeasured, and it names which box to buy a second of.""" + _p, _l, s = _sched() + assert s.bottleneck is not None + assert s.bottleneck.instrument == "namocell" + assert s.bottleneck.steps == 7 + # Under one-process-per-instrument those serialize across every plate in flight. + assert s.bottleneck.serialized_slots(10) == 70 + + +def test_contention_is_ranked_and_sums_to_the_protocol(): + p = protocols.get("single_cell_genomics") + rows = contention(p) + assert sum(r.steps for r in rows) == len(p.steps) + assert rows == sorted(rows, key=lambda c: (-c.steps, c.instrument)) + assert abs(sum(r.share for r in rows) - 1.0) < 1e-9 + + +def test_a_makespan_is_refused_however_many_plates_you_ask_for(): + """`throughput` refuses this for one plate. Adding plates makes the refusal stronger, + not weaker: a pipelined estimate compounds every unmeasured duration in it.""" + _p, _l, s = _sched(plates=500) + assert s.makespan() is None + assert "guess" in s.why_no_makespan() + # And the structural results are still there, which is the point of the design. + assert s.bottleneck.steps == 7 + + +# -- the purchase question ------------------------------------------------------ + + +def test_an_arm_removes_handoff_stalls_and_nothing_else(): + """The honest form of 'should we buy a plate mover'. It is the only thing that closes a + custody gap, and it decodes no command sets.""" + _p, _l, s = _sched() + removed, remains = s.arm_relief() + assert removed, "an arm should clear the physical hops" + assert all(st.constraint is Constraint.HANDOFF for st in removed) + assert not any(st.constraint is Constraint.HANDOFF for st in remains) + # It is not the next purchase here: far more stalls survive it than it clears. + assert len(remains) > len(removed) + assert any(st.constraint is Constraint.DECODE for st in remains) + + +def test_handoff_stalls_agree_with_the_ledgers_physical_hops(): + """One computation, not two -- the rule the provenance and lineage suites both enforce.""" + _p, ledger, s = _sched() + hops = [st for st in s.stalls if st.constraint is Constraint.HANDOFF] + assert len(hops) == len(ledger.handoffs()) == 5 + + +# -- classifying rather than collapsing ----------------------------------------- + + +def test_stalls_are_classified_because_each_one_costs_a_different_budget(): + """Collapsing these into 'blocked' is the same mistake as collapsing loop closure into + one autonomy percentage: it hides which budget to spend.""" + _p, _l, s = _sched() + kinds = s.by_constraint() + assert len(kinds) >= 3 + assert kinds["decode"] > 0 and kinds["handoff"] > 0 and kinds["attended"] > 0 + + +def test_every_stall_names_something_concrete_that_clears_it(): + """'Improve the workflow' is not an action.""" + _p, _l, s = _sched() + assert s.stalls + for st in s.stalls: + assert st.clears_it and len(st.clears_it) > 20 + assert st.detail + + +def test_where_the_run_stops_matches_the_ledgers_headless_prefix(): + """The scheduler must not invent a different answer to a question the ledger already + computed. Two numbers for 'how far does an unattended run get' would eventually + disagree and nothing would reveal which was wrong.""" + _p, ledger, s = _sched() + assert s.unattended_steps == ledger.headless_prefix() + + +# -- trust, scoped to where it actually binds ----------------------------------- + + +def test_a_read_only_probe_does_not_need_a_robot_benchmark(): + """Enumerating a USB bus moves nothing. Gating it on a volumetric-accuracy benchmark + would stall every protocol at step one and bury the constraints that differ.""" + _p, _l, s = _sched() + trust = [st for st in s.stalls if st.constraint is Constraint.TRUST] + assert not any(st.step_op == "discover_usb" for st in trust) + + +def test_trust_fires_where_the_machine_can_act_but_has_not_earned_it(): + """With plr-tested wired, two steps really run on hardware under supervision. Neither + has a met benchmark, and that -- not decoding -- is what stands between them and running + unattended.""" + _p, _l, s = _sched(plr_tested="/somewhere/plr-tested") + trust = {st.step_op for st in s.stalls if st.constraint is Constraint.TRUST} + assert "wgs_prep_lysis" in trust + assert all("no benchmark" in st.detail for st in s.stalls if st.constraint is Constraint.TRUST) + + +def test_wiring_in_real_hardware_moves_stalls_between_classes_not_away(): + """Proving run cards converts DECODE stalls into TRUST stalls. The run does not become + unattended; the reason it is not changes, and so does the next action.""" + _p, _l, bare = _sched() + _p2, _l2, wired = _sched(plr_tested="/somewhere/plr-tested") + assert wired.by_constraint().get("trust", 0) > bare.by_constraint().get("trust", 0) + assert wired.unattended_steps == bare.unattended_steps # still stops in the same place