Everyone means every entry of asked-of, and a case runs on its own clock

CHANGELOG.md

@@ -100,6 +100,23 @@ things were settled in the same exchange. Nothing here changes a field of the sc
   them nothing: it read the stage's `bind:`, which a waiting stage cannot write, and not the
   question's `shows:`, which a `decide` asking the same question did read. Found by the first
   shared case with a waiting stage in it.
+- **`everyone` means every entry of `asked-of:`, each answering separately.** The two cases for a
+  waiting stage merged in the entry above were wrong, and so was the reference: it took one answer
+  whatever `answered-by:` said, and the second case delivered one answer "from" two people, which
+  nothing could check. The schema's words are "how many of `asked-of:` must answer". Bud Agent
+  Flow, whose engine asks each entry, refused to build to the cases. Now: each entry is one
+  answerer (a person, or a team or channel one of whose members answers for it); all are asked at
+  once and each answers separately; one answer has one answerer; a second, different answer from
+  somebody who has answered, or one answer naming several people, is refused and recorded
+  (`refused` in a case's `expect`), changes nothing, and the wait stays open; the same answer
+  handed in again is a resend and counts once; one no is a no at once; the stage's fields are those of the answer that settled the wait (the
+  first no, or the answer that made it enough). Five cases replace the two (31 workflow cases, 55 cases of replayed
+  runs), on two workflows, `sign-off` and `sign-off-by-the-team`. In a case, a list under `people`
+  for a question is the answers as they arrive, and `shown` has one entry for each entry of
+  `asked-of:` that was asked.
+- **A shared case runs on its own clock.** The replay kit ran a case's first leg on the wall clock
+  and its resumes on the case's own, so a wait could never run out by itself. It starts at 0 now.
+  No case's expected result changed. (Bud Agent Flow found the same mixture in its replay.)
 - Ruled, with no case yet: a workflow run a port ended (`if-still-running: stop`,
   `cancel-previous`) ends as `cancelled`. `stopped` stays "could not go on, and says why".
 

README.md

@@ -1000,11 +1000,11 @@ framework is an extra.
 |---|---|
 | **Design** | Thesis, 29 binding decisions, FRD (120 requirements), 60-row refusal ledger — complete |
 | **Research** | 14 source-grounded studies, ~15,750 lines, over 140 repos (~15 GB) + 57 papers |
-| **Code** | Loader, diagnostics, schema engine, CLI, harness, resolver, evals, SLO — **5034 tests (1679 Rust + 3355 adapter), clippy clean, TypeScript type-checked** |
+| **Code** | Loader, diagnostics, schema engine, CLI, harness, resolver, evals, SLO — **5043 tests (1679 Rust + 3364 adapter), clippy clean, TypeScript type-checked** |
 | **Adapters** | **All 7 named targets**, proven against one shared conformance suite |
 
 ```bash
-./scripts/test-all.sh          # 5034 tests, Rust + 7 adapters, fully offline
+./scripts/test-all.sh          # 5043 tests, Rust + 7 adapters, fully offline
 ```
 
 **What is still owed** is not hidden — it lives in

adapters/python/src/pact_adapters/workflow_runtime.py

@@ -143,6 +143,15 @@ class WorkflowResult:
     #: its reason (`because:`), by stage: the run's "why this path". It is not
     #: part of the stage's record, which a binding reads.
     because: dict[str, str] = field(default_factory=dict)
+    #: Every time a question was put to somebody, in order: `{question,
+    #: shown}`, one for each entry of the question's `asked-of:`. A question
+    #: asked of two people is put twice, at once, whoever answers first.
+    asked: list[dict[str, Any]] = field(default_factory=list)
+    #: Answers that were handed in and did not count, with why
+    #: (`already-answered`: that person had answered this wait;
+    #: `several-people`: one answer named more than one answerer). Whoever
+    #: handed one in is told no. It is never taken as if it had counted.
+    refused: list[dict[str, Any]] = field(default_factory=list)
 
 
 #: Who answered a question, where the host says: a person's answer may carry
@@ -496,7 +505,9 @@ async def decide(self, stage: StageSpec, scope: _Scope) -> tuple[str, Any]:
             if rung.asks:
                 if standing is not None:
                     continue
-                answer = await self.person(rung.asks, self.shown_for(rung.asks, scope), stage.name)
+                shown = self.shown_for(rung.asks, scope)
+                self.put(rung.asks, shown, stage.name)
+                answer = await self.person(rung.asks, shown, stage.name)
                 who = answer.pop(ANSWERED_BY, None)
                 label = _one_of(answer, labels)
                 if label:
@@ -623,47 +634,133 @@ def decides_alone(self, agent: str) -> bool:
         rows = (self._catalogue.get(str(m).strip()) for m in models)
         return all(row is not None and row.kind == "decision" for row in rows)
 
+    def question(self, name: str) -> Mapping[str, Any]:
+        """The question of this name: the workflow's own, else the workspace's."""
+        mine = (self.document.get("workflows") or {}).get(self.spec.name) or {}
+        return (mine.get("questions") or {}).get(name) or (
+            (self.document.get("questions") or {}).get(name) or {}
+        )
+
     def shown_for(self, question: str, scope: _Scope) -> dict[str, Any]:
         """What a person who decides is shown: the question's own `shows:`
         lines, read — `shows: [input.message]`. It was `{}`, so the person was
         asked which desk should take a message nobody showed them."""
-        mine = (self.document.get("workflows") or {}).get(self.spec.name) or {}
-        asked = (mine.get("questions") or {}).get(question) or (
-            (self.document.get("questions") or {}).get(question) or {}
-        )
+        asked = self.question(question)
         shown: dict[str, Any] = {}
         for binding in asked.get("shows") or ():
             value = scope.read(str(binding))
             if value is not ABSENT:
                 shown[str(binding)] = value
         return shown
 
+    def put(self, question: str, shown: dict[str, Any], at: str) -> None:
+        """Put `question` to everybody it is `asked-of:`, and say so on the
+        result: once for each entry, because each entry is one answerer and
+        each is asked, whoever answers first. With nobody there to ask, the
+        run parks and nobody has been asked."""
+        if self.callers.person is None:
+            raise Parked(question, at)
+        entries = list(self.question(question).get("asked-of") or ()) or [None]
+        for _ in entries:
+            self.result.asked.append({"question": question, "shown": dict(shown)})
+
     async def person(self, question: str, shown: dict[str, Any], at: str) -> dict[str, Any]:
+        """One answer to a question that has been `put`."""
         if self.callers.person is None:
             raise Parked(question, at)
         return _as_answer(await _maybe(self.callers.person(question, shown)))
 
     async def ask(self, stage: StageSpec, scope: _Scope) -> tuple[str, Any]:
-        try:
-            # What the question's own `shows:` lists, as a `decide` asking the
-            # same question shows it. A stage that waits showed the person
-            # nothing: it passed its `bind:`, which a waiting stage cannot
-            # write, so "May this fix be closed?" arrived with no fix.
-            shown = {**self.shown_for(stage.asks, scope), **self.inputs(stage, scope)}
-            answer = await self.person(stage.asks, shown, stage.name)
-        except NobodyAnswered:
-            return "nobody-answered", {}
-        who = answer.pop(ANSWERED_BY, None)
-        if who:
-            # A LIST on a wait, always, in the order they answered: a question
-            # can need several answers (`answered-by: everyone`), and the
-            # shape of `steps.<stage>.answered-by` must not depend on how many
-            # people happened to answer. One name is a list of one. (On a
-            # `decide` it is one name: a decision is one person's.)
-            names = [str(w) for w in who] if isinstance(who, (list, tuple)) else [str(who)]
-            self.stage_words[id(stage)] = {ANSWERED_BY: names}
-        no = [k for k, v in answer.items() if isinstance(v, bool) and v is False]
-        return ("declined" if no else "answered"), answer
+        """A stage that waits for people.
+
+        `answered-by:` on the question says how many of `asked-of:` must
+        answer: the first (unwritten, or `anyone`), `enough-is:` of them, or
+        every one. The count is over the ENTRIES of `asked-of:`, each of which
+        is one answerer — a person, or a team or channel one of whose members
+        answers for it. So each is asked and answers separately, and one
+        answer has one answerer. This took a single answer whatever the
+        question said, and passed on whatever name or list of names a host
+        put beside it; the first shared cases for it then delivered one answer
+        "from" two people, which no runtime could check. Bud Agent Flow, whose
+        engine asks each entry, refused to build to them.
+
+        Everybody asked is asked at once (`put`), and the answers are taken
+        as they arrive. Any `no` is a no, at once: the rest are not waited
+        for. `steps.<stage>.answered-by` is who answered, in the order their
+        answers arrived, a list also when one person did.
+
+        An answer that does not count is REFUSED, and the wait stays open: a
+        second, different answer from somebody who has answered this wait
+        (nobody is "everyone" by answering for somebody else as well, and
+        nobody changes their answer by answering again), or one answer naming
+        several people. It is on the result as refused; it is never quietly
+        taken as if it had counted, because whoever handed it in would
+        believe it had.
+
+        The SAME answer from the same person, handed in again, is not a second
+        answer. Hosts resend: a file of answers passed whole at every resume,
+        a retry after a timeout. It counts once and nothing is refused.
+
+        The stage's fields are those of the answer that SETTLED the wait: the
+        first no, or the answer that made it enough. Several people can each
+        give a reason beside their yes, and one value has to be the stage's;
+        the one that settled it is the one the outcome came from.
+        """
+        asked = self.question(stage.asks)
+        entries = [str(e) for e in asked.get("asked-of") or ()]
+        needs = str(asked.get("answered-by") or "anyone").strip()
+        if needs == "everyone":
+            wanted = max(len(entries), 1)
+        elif needs == "enough-of-them":
+            wanted = max(int(asked.get("enough-is") or 1), 1)
+        else:
+            wanted = 1
+        # What the question's own `shows:` lists, as a `decide` asking the
+        # same question shows it. A stage that waits showed the person
+        # nothing: it passed its `bind:`, which a waiting stage cannot
+        # write, so "May this fix be closed?" arrived with no fix.
+        shown = {**self.shown_for(stage.asks, scope), **self.inputs(stage, scope)}
+        self.put(stage.asks, shown, stage.name)
+        names: list[str] = []
+        gave: dict[str, dict[str, Any]] = {}
+        fields: dict[str, Any] = {}
+        heard = 0
+        # Bounded, so a host that only ever hands in answers that do not
+        # count ends the run instead of spinning.
+        for _ in range(wanted + max(len(entries), 1) + 1):
+            if heard >= wanted:
+                break
+            try:
+                answer = await self.person(stage.asks, shown, stage.name)
+            except NobodyAnswered:
+                return "nobody-answered", {}
+            who = answer.pop(ANSWERED_BY, None)
+            if isinstance(who, (list, tuple)):
+                self.result.refused.append(
+                    {"question": stage.asks, "answered-by": [str(w) for w in who], "why": "several-people"}
+                )
+                continue
+            if who and str(who) in names:
+                if answer != gave[str(who)]:
+                    self.result.refused.append(
+                        {"question": stage.asks, "answered-by": str(who), "why": "already-answered"}
+                    )
+                continue
+            if who:
+                names.append(str(who))
+                gave[str(who)] = dict(answer)
+                self.stage_words[id(stage)] = {ANSWERED_BY: list(names)}
+            heard += 1
+            fields = answer
+            if any(isinstance(v, bool) and v is False for v in answer.values()):
+                return "declined", answer
+        if heard < wanted:
+            raise Stopped(
+                f"{stage.name!r} needs {wanted} of `asked-of:` to answer {stage.asks!r} and "
+                f"heard {heard}: every other answer handed in came from somebody who had "
+                f"answered already, or was refused ({len(self.result.refused)} refused)"
+            )
+        return "answered", fields
 
     async def each(self, stage: StageSpec, scope: _Scope, where: str) -> dict[str, Any]:
         items = (scope.read(stage.over))

adapters/python/tests/conformance_kit.py

@@ -32,6 +32,11 @@
 that was told, which is one for each yes, the first request after it and no
 other.
 
+* **The clock** — a case runs on its own clock and never on the wall clock:
+  the first leg at 0 seconds, each resume at the number of parks so far, and a
+  `"nobody"` resume one second after the wait has run out. A replay on the
+  wall clock is a different run each day: a wait with `answer-within:` is
+  inside its time today and past it tomorrow.
 * **`people`** — one entry for each time the run parks, in order: `"yes"` or
   `"no"`, given to every question of that park. `"nobody"` means no answer is
   given and the time the question allows (`answer-within:`) passes: the run is
@@ -126,7 +131,12 @@ async def model_call(self, system, history, tools):  # type: ignore[override]
         kw["checking_transport"] = ReferenceTransport(Script([Turn(str(case["judge"]))]))
 
     parks: list[dict[str, Any]] = []
-    out = asyncio.run(run(spec, Lead(lead), case["says"], tools, **kw))
+    # The case's own clock, from 0. This leg ran on the wall clock while every
+    # resume below ran on the case's, so a park was stamped with today's time
+    # and resumed at second 1: a wait could then never run out. No case's
+    # result depended on it. Bud Agent Flow found the same mixture in its own
+    # replay, where it did change a result.
+    out = asyncio.run(run(spec, Lead(lead), case["says"], tools, now=0.0, **kw))
     for said in case.get("people") or ():
         if out.halted != "suspended":
             break
@@ -178,7 +188,7 @@ def replay_workflow(case: dict[str, Any]) -> dict[str, Any]:
     doc = shown(REPO / case["tree"])
     spec = WorkflowSpec.from_workflow(doc, case["workflow"])
     given: list[dict[str, Any]] = []
-    shown_to: list[dict[str, Any]] = []
+    handed: dict[str, int] = {}
     called: list[dict[str, Any]] = []
 
     def agent(name: str, inputs: dict[str, Any]) -> dict[str, Any]:
@@ -194,15 +204,19 @@ def tool(name: str, action: str, inputs: dict[str, Any]) -> dict[str, Any]:
         return dict((case.get("tools") or {}).get(f"{name}/{action}", {}))
 
     def person(question: str, shown_: dict[str, Any]) -> dict[str, Any]:
-        shown_to.append({"question": question, "shown": dict(shown_)})
-        return dict((case.get("people") or {}).get(question, {}))
+        handed[question] = handed.get(question, 0) + 1
+        says = (case.get("people") or {}).get(question, {})
+        if isinstance(says, list):  # the answers as they arrive, the last repeating
+            says = says[min(handed[question], len(says)) - 1]
+        return dict(says)
 
     callers = Callers(agent=agent, tool=tool, person=person if case.get("people") is not None else None)
     out = asyncio.run(run_workflow(spec, case["input"], callers, document=doc))
     return {
         "trace": list(out.trace),
         "given": given,
-        "shown": shown_to,
+        "shown": list(out.asked),
+        "refused": list(out.refused),
         "called": called,
         "answer": json.loads(json.dumps(out.answer, default=str)),
         "halted": out.halted,

adapters/python/tests/test_the_shared_cases_are_what_this_runtime_does.py

@@ -113,15 +113,45 @@ def test_the_cases_say_the_things_they_are_there_to_say() -> None:
     asked = by_name["the-agent-is-not-sure-so-a-person-decides-and-is-shown-the-message"]
     assert asked["shown"] == [{"question": "which-desk", "shown": {"input.message": "The app crashes."}}]
 
-    one = by_name["who-answered-a-wait-is-a-list-of-one-when-one-person-did"]
-    # On a stage that waits for people, who answered is a list, also of one;
-    # and the person is shown what the question says to show.
-    assert one["answer"] == {"close": True, "by": ["dana"]}
-    assert one["shown"] == [
-        {"question": "may-we-close", "shown": {"input.report": "Seat 14C will not recline."}}
+    team = by_name["a-team-is-asked-and-who-answered-is-a-list-of-the-one-who-did"]
+    # A team is one entry of `asked-of:`. The first answer decides; who in the
+    # team gave it is a list of one, and not the name that was asked. The
+    # person is shown what the question says to show.
+    assert team["answer"] == {"close": True, "by": ["dana"]}
+    assert team["shown"] == [
+        {"question": "may-the-team-close", "shown": {"input.report": "Seat 14C will not recline."}}
     ]
-    both = by_name["who-answered-a-wait-is-everyone-who-did-in-the-order-they-answered"]
-    assert both["answer"]["by"] == ["lee", "dana"], "in the order they answered, not sorted"
+    both = by_name["everyone-asked-answers-separately-and-is-listed-in-the-order-they-answered"]
+    # `asked-of: [lee, dana]`, `answered-by: everyone`: both are asked, at
+    # once, and their answers are listed as they arrived, which is not the
+    # order they are written in.
+    assert len(both["shown"]) == 2 and both["answer"]["by"] == ["dana", "lee"]
+    assert both["refused"] == []
+    assert both["answer"]["outcome"] == "answered"
+    # Each gave a note. The stage's is the note of the answer that settled it:
+    # here the last yes that was needed, lee's.
+    assert both["answer"]["note"] == "Parts are logged."
+    said_no = by_name["one-no-from-everyone-asked-is-a-no-and-the-rest-are-not-waited-for"]
+    assert said_no["answer"] == {
+        "close": False, "by": ["dana"], "outcome": "declined",
+        "note": "The recline cable is still loose.",
+    }
+    # Both were asked. dana's no ended it, and lee's answer was not waited for.
+    assert len(said_no["shown"]) == 2
+    twice = by_name[
+        "a-second-answer-from-somebody-who-has-answered-is-refused-and-the-wait-stays-open"
+    ]
+    # dana said yes, then handed in a no: the second is refused, on the
+    # record, and changes nothing. The wait stays open until lee answers, and
+    # the answer is yes.
+    assert twice["refused"] == [
+        {"question": "may-we-close", "answered-by": "dana", "why": "already-answered"}
+    ]
+    assert twice["answer"]["by"] == ["dana", "lee"] and len(twice["shown"]) == 2
+    assert twice["answer"]["close"] is True and twice["answer"]["outcome"] == "answered"
+    resent = by_name["the-same-answer-handed-in-again-is-not-a-second-answer"]
+    # The same answer from dana again is a resend. Nothing is refused.
+    assert resent["refused"] == [] and resent["answer"]["by"] == ["dana", "lee"]
     found = by_name["a-deciding-agent-is-given-what-an-earlier-stage-found-and-the-bound-name-wins"]
     # `kind` is triage's `engine`, not the `cabin` the report came in with; the
     # `secret` the assigner does not accept is not given to it.