Plain Python with the standard library only: the cards, the service, the three schedules and the assistant’s rules as above. It runs as written with Python 3, prints the end of each schedule, and checks with assert that the assistant’s state is identical in A, B and C through step 11. The page itself runs a TypeScript version of the same rules, which also rejects malformed identifiers and payloads that these fixed schedules never contain.
import copy
# Snapshot cards-v1: harness, sample early recall in hundredths (94 means 0.94), role.
# Sample values, not measurements or trial counts.
SNAPSHOT = "cards-v1"
CARDS = {"base-0925": ("v1", 86, "baseline"),
"base-0928": ("v2", 94, "baseline"),
"int8-kv": ("v2", 93, "candidate")}
DECLARED = {"metric": "early-recall", "direction": "higher-is-better", "unit": "hundredths of recall",
"inclusive": True, "method": "difference-v1", "use": "next-test-screen"}
HIDDEN = ("service-receives", "request-lost", "service-reads") # the assistant never sees these
class Rejected(Exception):
"""A rejected event changes nothing: apply() works on a copy and drops it."""
def need(condition, reason):
if not condition:
raise Rejected(reason)
def compare(baseline, candidate):
"""Candidate minus baseline in hundredths, and whether this world allows the comparison."""
(b_harness, b_recall, _), (c_harness, c_recall, _) = CARDS[baseline[0]], CARDS[candidate[0]]
return c_recall - b_recall, b_harness == c_harness
def meets(difference, max_drop):
return difference >= -max_drop # inclusive: a drop of exactly max_drop passes
def payload_for(goal, baseline, candidate):
"""Everything an operation key names. Attempt IDs, times and narration are not part of it."""
return {"goal": goal["revision"], "baseline": baseline, "candidate": candidate,
"max_drop": goal["max_drop"], **DECLARED}
# ---- The service: the author's world. Operation key -> one immutable record. ----
def problem(p):
"""The service contract. None when the payload is admissible."""
if not all(ref[1] == SNAPSHOT and ref[0] in CARDS for ref in (p["baseline"], p["candidate"])):
return "unknown-card"
if (CARDS[p["baseline"][0]][2], CARDS[p["candidate"][0]][2]) != ("baseline", "candidate"):
return "roles"
if not compare(p["baseline"], p["candidate"])[1]:
return "not-comparable"
return None if all(p[k] == v for k, v in DECLARED.items()) else "undeclared"
def create_comparison(records, key, p):
"""Register the key and create its record in one step, or change nothing.
An empty map has nothing to compare with, so it cannot catch an unseen earlier meaning of a key."""
if key in records:
stored = records[key]
return records, (("existing", stored) if stored["payload"] == p else ("mismatch", key))
if problem(p):
return records, ("invalid", key)
difference = compare(p["baseline"], p["candidate"])[0]
record = {"id": "C" + key[1:], "key": key, "payload": p,
"difference": difference, "met": meets(difference, p["max_drop"])}
return {**records, key: record}, ("created", record)
def lookup(records, key):
"""What the map holds when the service reads it. Absent reserves and cancels nothing."""
return ("completed", records[key]) if key in records else ("absent", key)
def start():
world = {"records": {}, "requests": {}, "replies": {}, "lookups": {}, "answers": {}, "delivered": []}
actor = {"clock": 0, "goal": None, "goals": [], "baseline": None, "candidate": ("int8-kv", SNAPSHOT),
"asked": 0, "bound": {}, "attempts": {}, "lookups": {}, "completed": {},
"absent": [], "refused": [], "decisions": [], "said": [], "deliveries": []}
return {"world": world, "actor": actor}
def world_event(W, e):
if e["type"] == "service-reads":
read = W["lookups"].get(e["lookup"])
need(read is not None and read["status"] == "in-transit", "not-in-transit")
read["status"] = "read"
W["answers"][e["lookup"]] = lookup(W["records"], read["key"])
return
request = W["requests"].get(e["attempt"])
need(request is not None and request["status"] == "in-transit", "not-in-transit")
if e["type"] == "request-lost":
request["status"] = "lost"
else:
request["status"] = "received"
W["records"], W["replies"][e["attempt"]] = create_comparison(W["records"], request["key"], request["payload"])
def receive(S, key, record, t, via):
"""Accept a record only if it is the one this key was bound to. Count each record once.
Returns False when the reply is not evidence for this key."""
if record["key"] != key or S["bound"].get(key) != record["payload"]:
S["deliveries"].append((t, via, "rejected-evidence", None))
return False
if record["id"] in S["completed"]:
S["deliveries"].append((t, via, "duplicate", record["id"]))
else:
S["completed"][record["id"]] = {**record, "seen": t}
S["deliveries"].append((t, via, "new", record["id"]))
return True
def allowance(x):
return type(x) is int and 0 <= x <= 100 # hundredths of recall
def apply(state, e):
state = copy.deepcopy(state)
W, S, kind = state["world"], state["actor"], e["type"]
if kind in HIDDEN:
world_event(W, e)
return state
S["clock"] += 1 # counts only what the assistant does or receives
t, goal = S["clock"], S["goal"]
if kind == "person-asks":
need(goal is None, "goal-already-set")
need(allowance(e["max_drop"]), "criterion-out-of-range")
S["goal"] = {"revision": 1, "max_drop": e["max_drop"]}
S["goals"], S["asked"] = [S["goal"]], 1
elif kind == "assistant-says": # commentary: not an observation, changes no status
need(e["text"].strip(), "empty-statement")
S["said"].append((t, e["text"]))
elif kind == "person-selects-baseline":
ref = e["baseline"]
need(ref[1] == SNAPSHOT and ref[0] in CARDS, "unknown-card")
need(CARDS[ref[0]][2] == "baseline", "not-a-baseline")
S["baseline"] = ref
elif kind == "assistant-calculates": # needs only the cards, never the service
need(goal is not None, "goal-not-set")
need(S["baseline"] is not None, "reference-unresolved")
difference, comparable = compare(S["baseline"], S["candidate"])
S["decisions"].append({"goal": goal["revision"], "baseline": S["baseline"], "difference": difference,
"met": meets(difference, goal["max_drop"]) if comparable else None})
elif kind == "assistant-prepares":
need(goal is not None, "goal-not-set")
need(S["baseline"] is not None, "reference-unresolved")
need(compare(S["baseline"], S["candidate"])[1], "not-comparable")
need(e["key"] not in S["bound"], "key-already-bound") # bound once, before the service sees it
need(S["asked"] > len(S["bound"]), "no-open-request") # a timeout is not a request
S["bound"][e["key"]] = payload_for(goal, S["baseline"], S["candidate"])
elif kind == "assistant-sends":
key, attempt = e["key"], e["attempt"]
need(key in S["bound"], "unbound-key")
need(attempt not in S["attempts"], "attempt-reused")
need(applies(S, S["bound"][key]), "obsolete-request") # earlier criterion or baseline
need(all(c["key"] != key for c in S["completed"].values()), "already-completed")
need(all(r[0] != key for r in S["refused"]), "already-refused")
S["attempts"][attempt] = {"key": key, "sent": t, "stopped": None, "reply": None}
W["requests"][attempt] = {"key": key, "payload": S["bound"][key], "status": "in-transit"}
elif kind == "assistant-stops-waiting": # the assistant's own clock, not an answer
a = S["attempts"].get(e["attempt"])
need(a is not None, "unknown-attempt")
need(a["reply"] is None, "not-waiting")
need(a["stopped"] is None, "already-stopped-waiting")
a["stopped"] = t
elif kind == "person-changes-criterion": # a new revision; it cancels nothing already sent
need(goal is not None, "goal-not-set")
need(allowance(e["max_drop"]), "criterion-out-of-range")
need(e["max_drop"] != goal["max_drop"], "criterion-unchanged")
S["goal"] = {"revision": len(S["goals"]) + 1, "max_drop": e["max_drop"]}
S["goals"].append(S["goal"])
elif kind == "assistant-looks-up":
need(e["key"] in S["bound"], "unbound-key")
need(e["lookup"] not in S["lookups"], "lookup-reused")
S["lookups"][e["lookup"]] = {"key": e["key"], "sent": t, "answered": None}
W["lookups"][e["lookup"]] = {"key": e["key"], "status": "in-transit"}
elif kind == "person-requests-record":
need(goal is not None, "goal-not-set")
need(S["asked"] <= len(S["bound"]), "request-already-open")
S["asked"] += 1
elif kind == "reply-delivered":
need(e["delivery"] not in W["delivered"], "delivery-reused")
need(e["attempt"] in W["replies"], "no-reply-yet")
need(e["attempt"] in S["attempts"], "unknown-attempt")
W["delivered"].append(e["delivery"])
a, outcome = S["attempts"][e["attempt"]], W["replies"][e["attempt"]]
if outcome[0] in ("created", "existing"):
evidence = receive(S, a["key"], outcome[1], t, e["delivery"])
else:
evidence, known = True, (a["key"], outcome[0]) in S["refused"]
S["refused"] += [] if known else [(a["key"], outcome[0])]
S["deliveries"].append((t, e["delivery"], "duplicate" if known else "refusal", None))
if evidence: # a reply that is not evidence does not end the wait
a["reply"] = a["reply"] or t
elif kind == "answer-delivered":
need(e["delivery"] not in W["delivered"], "delivery-reused")
need(e["lookup"] in W["answers"], "no-answer-yet")
need(e["lookup"] in S["lookups"], "unknown-lookup")
W["delivered"].append(e["delivery"])
read, answer = S["lookups"][e["lookup"]], W["answers"][e["lookup"]]
if answer[0] == "completed":
evidence = receive(S, read["key"], answer[1], t, e["delivery"])
else: # absence at one reading, kept beside any completion
evidence, known = True, any(lk == e["lookup"] for _, lk, _ in S["absent"])
S["absent"] += [] if known else [(read["key"], e["lookup"], t)]
S["deliveries"].append((t, e["delivery"], "duplicate" if known else "absent", None))
if evidence:
read["answered"] = read["answered"] or t
else:
raise Rejected("unknown-event")
return state
# ---- What the assistant can say from its own state ----
def number(name):
return int(name[1:])
def status(S, key):
record = next((c for c in S["completed"].values() if c["key"] == key), None)
sent = [a["sent"] for a in S["attempts"].values() if a["key"] == key]
absent = [t for k, _, t in S["absent"] if k == key]
if record:
return "completed", record
if any(k == key for k, _ in S["refused"]):
return "refused", None
if max(absent, default=0) > max(sent, default=0):
return "absent-at-lookup", None
return ("sent" if sent else "prepared"), None
def applies(S, p):
"""Same goal revision, same references, stated method: it answers the current question."""
goal = S["goal"]
return (goal is not None and p["goal"] == goal["revision"] and p["max_drop"] == goal["max_drop"]
and p["baseline"] == S["baseline"] and p["candidate"] == S["candidate"] and problem(p) is None)
def summary(state):
W, S = state["world"], state["actor"]
goal, seen = S["goal"], sorted(S["completed"], key=number)
operations = []
for key in sorted(S["bound"], key=number):
state_of, record = status(S, key)
stopped = any(a["key"] == key and a["stopped"] and not a["reply"] for a in S["attempts"].values())
operations.append([key, state_of, stopped, record["id"] if record else None])
current = [d for d in S["decisions"] if goal and d["met"] is not None
and d["goal"] == goal["revision"] and d["baseline"] == S["baseline"]]
return {"goal": [goal["revision"], goal["max_drop"]] if goal else None,
"baseline": S["baseline"][0] if S["baseline"] else None,
"operations": operations,
"observed": seen,
"fromCards": [current[-1]["goal"], current[-1]["difference"], current[-1]["met"]] if current else None,
"saved": [r for r in seen if applies(S, S["completed"][r]["payload"])],
"history": [r for r in seen if not applies(S, S["completed"][r]["payload"])],
"world": [[k, r["id"], r["payload"]["goal"], r["met"]] for k, r in sorted(W["records"].items(), key=lambda kv: number(kv[0]))],
"inTransit": sorted((a for a, m in W["requests"].items() if m["status"] == "in-transit"), key=number),
"deliveries": [list(d) for d in S["deliveries"]],
"absences": [list(a) for a in S["absent"]],
"refusals": [list(r) for r in S["refused"]]}
# ---- Three delivery schedules, fixed before anything happens ----
def ev(kind, **fields):
return {"type": kind, **fields}
OPENING = [ev("person-asks", max_drop=2),
ev("assistant-says", text="Two baseline cards are visible: base-0925, scored on harness v1, "
"and base-0928, on v2. Which one do you mean?"),
ev("person-selects-baseline", baseline=("base-0928", SNAPSHOT)),
ev("assistant-calculates"),
ev("assistant-prepares", key="K1"),
ev("assistant-sends", key="K1", attempt="A1")]
MIDDLE = [ev("assistant-stops-waiting", attempt="A1"),
ev("assistant-says", text="The comparison failed. I will submit it again."),
ev("person-changes-criterion", max_drop=0),
ev("assistant-calculates"),
ev("assistant-looks-up", key="K1", lookup="L1"),
ev("service-reads", lookup="L1"),
ev("answer-delivered", lookup="L1", delivery="d1"),
ev("person-requests-record"),
ev("assistant-prepares", key="K2"),
ev("assistant-sends", key="K2", attempt="A2"),
ev("service-receives", attempt="A2"),
ev("reply-delivered", attempt="A2", delivery="d2")]
LATE_TWICE = [ev("reply-delivered", attempt="A1", delivery="d3"),
ev("reply-delivered", attempt="A1", delivery="d4")]
SCHEDULES = {"A": OPENING + [ev("service-receives", attempt="A1")] + MIDDLE + LATE_TWICE, # saved; reply held
"B": OPENING + [ev("request-lost", attempt="A1")] + MIDDLE, # lost on the way
"C": OPENING + MIDDLE + [ev("service-receives", attempt="A1")] + LATE_TWICE} # arrives late
def run(events, state=None):
"""The state after each event the assistant takes part in or receives."""
state, states = state or start(), []
for e in events:
state = apply(state, e)
if e["type"] not in HIDDEN:
states.append(state)
return states
def words(names):
return " ".join(names) or "none"
runs = {name: run(events) for name, events in SCHEDULES.items()}
for name, states in runs.items():
s = summary(states[-1])
g, d, met = s["fromCards"]
print(f"{name}: service holds {words(r[1] for r in s['world'])}; assistant observed {words(s['observed'])}")
print(" " + ", ".join(f"{k} {st}" for k, st, _, _ in s["operations"]))
print(f" answer for G{g} from the cards: {d / 100:+.2f}, {'met' if met else 'not met'}; "
f"saved for G{g}: {words(s['saved'])}; history: {words(s['history'])}")
# Through step 11 the assistant's state is identical in A, B and C, whatever the service did.
assert [s["actor"] for s in runs["A"][:11]] == [s["actor"] for s in runs["B"][:11]] == [s["actor"] for s in runs["C"][:11]]
assert summary(runs["A"][6])["world"] != summary(runs["B"][6])["world"]
# A delivered twice is still one completion; C2 completed although its criterion is not met.
assert summary(runs["A"][-1])["observed"] == ["C1", "C2"] and not runs["A"][-1]["world"]["records"]["K2"]["met"]
# After the absent lookup in B, K1 still names only its G1 payload.
try:
apply(runs["B"][11], ev("assistant-prepares", key="K1"))
except Rejected as why:
print("rebinding K1 after an absent lookup:", why)
# Restoring the 0.02 wording is revision G3: C1 stays history; a fresh answer comes from the cards.
r = summary(run([ev("person-changes-criterion", max_drop=2), ev("assistant-calculates")], runs["A"][-1])[-1])
print(f"restored wording is G{r['goal'][0]}: from the cards {r['fromCards'][1] / 100:+.2f}, "
f"{'met' if r['fromCards'][2] else 'not met'}; saved: {words(r['saved'])}; history: {words(r['history'])}")
It prints:
A: service holds C1 C2; assistant observed C1 C2
K1 completed, K2 completed
answer for G2 from the cards: -0.01, not met; saved for G2: C2; history: C1
B: service holds C2; assistant observed C2
K1 absent-at-lookup, K2 completed
answer for G2 from the cards: -0.01, not met; saved for G2: C2; history: none
C: service holds C1 C2; assistant observed C1 C2
K1 completed, K2 completed
answer for G2 from the cards: -0.01, not met; saved for G2: C2; history: C1
rebinding K1 after an absent lookup: key-already-bound
restored wording is G3: from the cards -0.01, met; saved: none; history: C1 C2