coulomb/mail-address-semantics
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MAS-WP-0013 finished: mode contract flows runnable from the bundle alone

Browse source 
conformance/FLOWS.md defines setup, steps, captures and expectations, all
observable at the contract boundary. flows.json carries setups (peer ids),
fixed test keys, statute, scenario and explicit assign steps;
scenarios.json holds the callback scenarios (the stub loads them).
CONTRACT.md defines the case.assign response. Runner reads only the
bundle and rejects undefined keys; new flow s-006 (held over cap).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5787@bnt-lap001
Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd
This commit is contained in:
tegwick 2026-10-19 12:49:15 +02:00
parent 8de8fddf37
commit ecb1ba47f6
14 changed files with 1118 additions and 269 deletions
Download patch file Download diff file Expand all files Collapse all files

46
mas_modes/stubs/callback.py
View file

@ -18,32 +18,32 @@ import threading
from dataclasses import dataclass, field
from typing import Optional
from .. import message_errors
from .. import _load, message_errors
from ..contract import TERMINAL, Idempotent, json_server, message, post, refuse
AT = "2026-10-20T10:00:00Z"
NEEDED = ("phone", "time")
WAIT = "wait-for-details" # negotiation: block until handoff.update supplies NEEDED
WAIT = "wait-for-details" # negotiation: block until handoff.update supplies the needed fields
# Each step is (status, extra fields), or WAIT.
SCENARIOS: dict[str, list] = {
"completed": [("scheduled", {"at": AT}), ("running", {}), ("completed", {})],
"negotiate": [WAIT, ("scheduled", {"at": AT}), ("running", {}), ("completed", {})],
"no-answer-retry": [("scheduled", {"at": AT}), ("attempted", {"outcome": "no-answer"}),
("scheduled", {"at": "2026-10-20T14:00:00Z"}), ("running", {}),
("completed", {})],
"no-answer-fail": [("scheduled", {"at": AT}), ("attempted", {"outcome": "no-answer"}),
("scheduled", {"at": "2026-10-20T14:00:00Z"}),
("attempted", {"outcome": "no-answer"}),
("failed", {"detail": "two attempts, no answer"})],
"call-failed": [("scheduled", {"at": AT}), ("running", {}),
("failed", {"detail": "line dropped"})],
"expired": [("scheduled", {"at": AT}), ("expired", {"detail": "slot passed"})],
"unavailable": [("refused", {"reason": "mode-unavailable"})],
"declined": [("refused", {"reason": "declined"})],
"out-of-hours": [("refused", {"reason": "out-of-hours"})],
"silent": [], # never answers: the timeout owner has to act
}
def _scenarios() -> tuple[dict, tuple]:
"""The scenarios from the bundle (conformance/scenarios.json), as steps of
(status, extra fields) or WAIT."""
doc = _load("conformance/scenarios.json")
needed = ()
out = {}
for name, steps in doc["scenarios"].items():
script = []
for step in steps:
if "wait_for_details" in step:
needed = tuple(step["wait_for_details"])
script.append(WAIT)
else:
extra = {k: v for k, v in step.items() if k != "status"}
script.append((step["status"], extra))
out[name] = script
return out, needed
SCENARIOS, NEEDED = _scenarios()
@dataclass