tegwick
71c17fb5f9
impl/js/src/email.mjs (written from CONTRACT.md + FLOWS.md): both roles, case store, handoffs, timeouts, one-cap outbox with holding, authorization. Passes all 22 flows from the bundle; with the JS master all 12 slave flows (no Python contract party). Writing it found six contract gaps, folded into the draft as MCP-1..6 (acknowledgement, details convention, resume names the outcome, opaque case key, error table, transfer reporting); flows now check outcome tokens, not reference prose. A mutation check found a missing flow: a-008 (per-type master check). 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
109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""An outside master in another process, driven through its control API.
|
|
|
|
Same interface as `stubs.master.StubMaster`, so the harness and the flow
|
|
runner can swap it in. Used for the independent JS master in
|
|
`impl/js/src/master.mjs` (MAS-WP-0009).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
JS_MASTER = ROOT / "impl/js/src/master.mjs"
|
|
|
|
|
|
def _call(url: str, body: Optional[dict] = None):
|
|
data = None if body is None else json.dumps(body).encode()
|
|
req = urllib.request.Request(url, data=data, method="GET" if body is None else "POST",
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
return json.loads(r.read().decode())
|
|
|
|
|
|
class ExternalMaster:
|
|
def __init__(self, argv: list[str]):
|
|
self.proc = subprocess.Popen(argv, stdout=subprocess.PIPE, text=True)
|
|
info = json.loads(self.proc.stdout.readline())
|
|
self.events_url, self.control_url = info["events_url"], info["control_url"]
|
|
self._email_url = None
|
|
|
|
@classmethod
|
|
def js(cls, key: bytes, peer_id: str = "cms-1") -> "ExternalMaster":
|
|
return cls(["node", str(JS_MASTER), "--peer-id", peer_id, "--key-hex", key.hex()])
|
|
|
|
@property
|
|
def email_url(self):
|
|
return self._email_url
|
|
|
|
@email_url.setter
|
|
def email_url(self, url):
|
|
self._email_url = url
|
|
_call(self.control_url + "/config", {"email_url": url})
|
|
|
|
@property
|
|
def received(self) -> list[dict]:
|
|
return _call(self.control_url + "/received")
|
|
|
|
def of_type(self, type_: str) -> list[dict]:
|
|
return [m for m in self.received if m["type"] == type_]
|
|
|
|
def command(self, type_: str, case: Optional[dict] = None, **fields):
|
|
out = _call(self.control_url + "/command",
|
|
{"type": type_, **({"case": case} if case else {}),
|
|
**{k: v for k, v in fields.items() if v is not None}})
|
|
return out["status"], out["body"]
|
|
|
|
def assign(self, intake_ref: str, resource_id: str):
|
|
return self.command("case.assign", intake_ref=intake_ref, resource_id=resource_id)
|
|
|
|
def policy(self, case: dict, autonomous_send=None, transfers=None):
|
|
return self.command("case.policy", case, autonomous_send=autonomous_send, transfers=transfers)
|
|
|
|
def send(self, case: dict, in_reply_to: str, body: str, subject: Optional[str] = None):
|
|
return self.command("mail.send", case, in_reply_to=in_reply_to, body=body, subject=subject)
|
|
|
|
def cancel(self, case: dict, handoff_id: str, reason: str = "timed-out"):
|
|
return self.command("handoff.cancel", case, handoff_id=handoff_id, reason=reason)
|
|
|
|
def pump(self):
|
|
return []
|
|
|
|
def close(self):
|
|
self.proc.terminate()
|
|
self.proc.wait(timeout=10)
|
|
|
|
|
|
JS_EMAIL = ROOT / "impl/js/src/email.mjs"
|
|
|
|
|
|
class ExternalEmail:
|
|
"""An email side in another process (e.g. impl/js/src/email.mjs), configured
|
|
with a flow setup and driven through its control API."""
|
|
|
|
def __init__(self, argv: list[str], config: dict, scratch: Path):
|
|
cfg = scratch / f"email-{id(self)}.json"
|
|
cfg.write_text(json.dumps(config))
|
|
self.proc = subprocess.Popen([*argv, "--config", str(cfg)], stdout=subprocess.PIPE, text=True)
|
|
self.base_url = json.loads(self.proc.stdout.readline())["base_url"]
|
|
|
|
@classmethod
|
|
def js(cls, config: dict, scratch: Path) -> "ExternalEmail":
|
|
return cls(["node", str(JS_EMAIL)], config, scratch)
|
|
|
|
def outbox(self) -> list[dict]:
|
|
return _call(self.base_url + "/control/outbox")
|
|
|
|
def tick(self) -> list[str]:
|
|
return self.advance(0)
|
|
|
|
def advance(self, seconds: float) -> list[str]:
|
|
return _call(self.base_url + "/control/advance", {"seconds": seconds})["timed_out"]
|
|
|
|
def close(self):
|
|
self.proc.terminate()
|
|
self.proc.wait(timeout=10)
|