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