Ticket graphs

Rewrite ONE unique forge module as entirely new, behavior-preserving, installable, control-config'd Python, per the FORGE-wide refactor standard #200 (#std). This ticket covers ONE module — the one identified below — and nothing else. It was filed against the module's DEDUPLICATED identity, so if that module was copy-pasted across the forge, every copy is covered by this single ticket. MODULE KEY : forge-rw:653bf1a78a5f85dc (the ticketer's idempotency marker — it lives in this ticket's TITLE. Do not edit the title, or the module will be re-filed as a duplicate.) IDENTITY : AST-NORMALIZED — codebean's `_ast_norm` (#183), i.e. ast.dump(ast.parse(ast.unparse(ast.parse(src)))). Copies of this module that differ only in comments, blank lines, quoting or spacing carry the SAME module-sha and are covered by THIS ticket. A differing docstring is real content, so a module whose docstring differs is a DIFFERENT module with its own ticket. SIZE : 20952 bytes EXTENSION : .py CANONICAL SOURCE (the location this ticket is named for): repo : agent_tools_suite path : ats_http.py ref : HEAD (HEAD) blob : 139ce250f685b1090711727eb393232f293a0c48 (git blob sha1) REFERENCE ONLY — you do NOT need this command. This module's bytes are embedded COMPLETE in THE ORIGINAL SOURCE block below, and that is what the rewrite is based on. For a reader who does have forge access: git --git-dir=/Users/stevenpeterson/code/installed/jazz-project/config/forge/remotes/agent_tools_suite.git cat-file blob 139ce250f685b1090711727eb393232f293a0c48 THE SAME MODULE ALSO LIVES AT THESE FORGE TIPS: none — this module is unique to agent_tools_suite among the repos scanned so far. HISTORICAL OCCURRENCES of this exact blob — every OTHER DISTINCT {repo, path} the omni-git index (`occ`, joined on the git blob sha1) records this byte-identical file at, with the index's own `ts` for the first row of each. Any place this run enumerated LIVE is listed under the forge tips above, not here. Context for the rewrite, not extra work: testmonkeyalpha-group/jazz-forge/agent_tools_suite:ats_http.py [occ.ts gitlab] THE ORIGINAL SOURCE, EMBEDDED — YOU NEED NO FORGE ACCESS TO DO THIS TICKET. The module's own bytes are reproduced below, inside this ticket. THEY are the authoritative copy and the behavior-preserving basis: rewrite from them. The `git --git-dir=... cat-file blob` command under CANONICAL SOURCE above is kept only as a REFERENCE for a reader who happens to have forge access — running it is NOT part of this ticket. (Why the bytes are here at all: the pair_loop worker that executes this ticket is sandboxed to the ticket's own congruency worktree and cannot read the forge's bare repos. Ticket #346 is the fix that put them on the ticket.) The block BEGINS at the line ===ORIGINAL SOURCE (verbatim, the behavior-preserving basis)=== and ENDS at the first line ===END SOURCE=== Everything strictly between those two lines is the module — 20952 byte(s), byte for byte, with no re-indentation, no re-wrapping and nothing elided. ===ORIGINAL SOURCE (verbatim, the behavior-preserving basis)=== #!/usr/bin/env python3 """ats_http — the agent tool suite as ONE FastMCP service over HTTP/SSE, with the gate's requirements enforced SERVER-SIDE so they can't be side-stepped. The old stdio design let a client talk to a leaf/coupler stdin directly and skip the gate (no intent, no memory ticket, no provenance). Here every call passes through a single override of `FastMCP.call_tool` -> `enforce()`: a required `intent`, a memory-ticket precondition on state changes, and a HARD provenance write to congruency (a state change that can't be recorded is refused, not silently dropped). Tool BODIES are the existing leaf logic, reused unchanged. Run: python3 ats_http.py # streamable-http on 127.0.0.1:8811 ATS_TRANSPORT=sse python3 ats_http.py Honest scope: this closes side-stepping AT THE MCP LAYER. It does NOT stop a shell from running mint_crank.py / install.py / wallet / :8899 directly — that needs capability tokens or VM confinement. """ import hashlib import importlib.util import json import logging import os import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime, timezone from logging.handlers import RotatingFileHandler from mcp.server.fastmcp import FastMCP from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.auth.provider import AccessToken, TokenVerifier from mcp.server.auth.settings import AuthSettings import ats_token HERE = os.path.dirname(os.path.abspath(__file__)) CONG = (os.environ.get("CONGRUENCY_URL") or "http://127.0.0.1:8899").rstrip("/") CRANKD = (os.environ.get("CRANKD_URL") or "http://127.0.0.1:8812").rstrip("/") # the crank service IMGD = (os.environ.get("IMGD_URL") or "http://127.0.0.1:8813").rstrip("/") # the pi-image service WORKSTATION = os.path.expanduser("~/code/workstation") # crankws_read is confined to this repo CONGRUENCY_REPO = os.path.expanduser("~/code/congruency") # crankcong_read is confined to this repo SESSION = os.environ.get("CLAUDE_CODE_SESSION_ID") or ("ats-http-%d" % os.getpid()) JSON_LOG = os.path.join(HERE, "ats-http-memories.jsonl") # local fallback provenance STATE_CHANGE = {"crankws_crank"} # tools that mutate real state LOG_PATH = os.environ.get("ATS_LOG") or os.path.expanduser("~/bin/config/ats/gateway.log") def _secret(name): """Read a value from ~/.secrets/ats.env (KEY=VALUE), else the process env; None if absent.""" path = os.path.expanduser("~/.secrets/ats.env") if os.path.isfile(path): for line in open(path): line = line.strip() if line.startswith(name + "="): return line.split("=", 1)[1] return os.environ.get(name) CONGRUENCY_KEY = _secret("CONGRUENCY_API_KEY") # X-Api-Key so congruency accepts the gate's writes def _setup_logging(): """Route every gate decision + the SDK's own logs to one rotating, greppable file. The official SDK's FastMCP loggers propagate to root, so a root handler captures them too.""" try: os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True) h = RotatingFileHandler(LOG_PATH, maxBytes=5_000_000, backupCount=5) h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")) root = logging.getLogger() root.setLevel(logging.INFO) if not any(getattr(x, "baseFilename", None) == h.baseFilename for x in root.handlers): root.addHandler(h) # idempotent across re-import logging.getLogger("mcp").setLevel(logging.INFO) except OSError: pass return logging.getLogger("ats") log = _setup_logging() def _leaf(rel, name): spec = importlib.util.spec_from_file_location(name, os.path.join(HERE, rel)) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) return m crankws = _leaf("crankws-mcp/server.py", "ats_crankws") # crankfix + crankimg REMOVED from the gate: a gated tool body must not read the db directly or spawn a # subprocess. Congruency work goes through its REST API; cranks through crankd; image ops through imgd. # ── provenance + ticket (server-side, the un-side-steppable requirement) ──────────────────────── def now_iso(): return datetime.now(timezone.utc).isoformat() def _actor(): """Who is making THIS call — the authenticated token's client_id (operator / crankws-agent), or the gate's own session when there is no request context. Provenance is stamped with this, so an agent is distinguishable from the operator (both used to inherit the gate process's session).""" tok = get_access_token() return tok.client_id if tok else SESSION def _cong(method, path, obj=None): data = json.dumps(obj).encode() if obj is not None else None headers = {"Content-Type": "application/json"} if CONGRUENCY_KEY: headers["X-Api-Key"] = CONGRUENCY_KEY # congruency refuses keyless writes req = urllib.request.Request(CONG + path, data=data, method=method, headers=headers) with urllib.request.urlopen(req, timeout=3) as r: # raises on failure — callers decide return json.loads(r.read().decode() or "{}") def record(tool, intent, note, ts): """Record a memory to congruency; return True on success. Always also append the local fallback. Stamped with the caller's actor (_actor) so the agent and operator are distinguishable in the feed.""" actor = _actor() try: with open(JSON_LOG, "a") as fh: fh.write(json.dumps({"ts": ts, "session": actor, "tool": tool, "intent": intent, "note": note}) + "\n") except OSError: pass try: _cong("POST", "/?api=memories", {"ts": ts, "session": actor, "tool": tool, "intent": intent, "note": note}) return True except Exception: # noqa: BLE001 return False def open_ticket(): try: rows = _cong("GET", "/?api=tickets&component=crank-workstation&status=OPEN&order=id.desc&per=1").get("rows") return bool(rows) except Exception: # noqa: BLE001 return False class GateRefusal(Exception): """A requirement of the MCP was not met — the call is refused before dispatch.""" def _tool_scopes(name): """Scopes that grant a tool: crankws_* need crankws-or-full; crankcong_* need crankcong-or-full; everything else needs full.""" if name.startswith("crankws_"): return {"ats:crankws", "ats:full"} if name.startswith("crankcong_"): return {"ats:crankcong", "ats:full"} return {"ats:full"} def current_scopes(): """The authenticated token's scopes for this request, or None when auth is disabled / there is no HTTP context (e.g. direct in-process tests) — None means 'allow all'.""" tok = get_access_token() return set(tok.scopes) if tok else None def enforce(name, arguments): """The single, un-side-steppable precondition for EVERY tool call. Raises GateRefusal to refuse. Directly testable (no HTTP context needed).""" args = arguments or {} intent = args.get("intent") ts = now_iso() if not intent: record(name, None, "REFUSED: no intent", ts) log.warning("REFUSE %s session=%s reason=no-intent", name, SESSION) raise GateRefusal("gate: 'intent' is required for %s (why are you calling it?)" % name) have = current_scopes() if have is not None and not (have & _tool_scopes(name)): record(name, intent, "REFUSED: scope", ts) log.warning("REFUSE %s session=%s reason=scope have=%s", name, SESSION, sorted(have)) raise GateRefusal("scope: this token is not authorized for %s (needs one of %s)" % (name, sorted(_tool_scopes(name)))) if name in STATE_CHANGE: # authorization for a crank now lives in crankd (wallet-signed ticket approval); the gate does # intent + scope + HARD provenance, then the tool body forwards the request to crankd. if not record(name, intent, None, ts): # HARD: a state change MUST be provenanced log.error("REFUSE %s session=%s intent=%r reason=provenance-unrecorded", name, SESSION, intent[:80]) raise GateRefusal("state change DISALLOWED: provenance could not be recorded (congruency unreachable)") log.info("ALLOW %s (state-change) session=%s intent=%r", name, SESSION, intent[:120]) else: record(name, intent, None, ts) # reads: best-effort provenance log.info("ALLOW %s session=%s intent=%r", name, SESSION, intent[:120]) return ts class GatedMCP(FastMCP): async def call_tool(self, name, arguments): enforce(name, arguments) # raises GateRefusal -> isError result, no dispatch return await super().call_tool(name, arguments) async def list_tools(self): tools = await super().list_tools() have = current_scopes() if have is None: # auth disabled / no context -> full surface return tools return [t for t in tools if _tool_scopes(t.name) & have] class WalletTokenVerifier(TokenVerifier): """Bearer verifier over the wallet-derived token map ({token: scopes}). Unknown token -> None (401).""" def __init__(self, token_map): self._map = token_map async def verify_token(self, token): scopes = self._map.get(token) if scopes is None: return None client = ("operator" if "ats:full" in scopes else "crankws-agent" if "ats:crankws" in scopes else "crankcong-agent" if "ats:crankcong" in scopes else "agent") return AccessToken(token=token, client_id=client, scopes=list(scopes), expires_at=None) # ── the service + the tools (bodies = the reused leaf logic; `intent` is required by every tool) ── _HOST = os.environ.get("ATS_HOST", "127.0.0.1") _PORT = int(os.environ.get("ATS_PORT", "8811")) _TOKENS = ats_token.load_map() # {token: scopes}; {} if no ats.env and no wallet if _TOKENS: _base = "http://%s:%d" % (_HOST, _PORT) _auth = AuthSettings(issuer_url=_base, resource_server_url=_base, required_scopes=["ats:use"]) mcp = GatedMCP("ats", host=_HOST, port=_PORT, token_verifier=WalletTokenVerifier(_TOKENS), auth=_auth) log.info("auth ENABLED — %d wallet tokens, base scope 'ats:use' required by transport", len(_TOKENS)) else: mcp = GatedMCP("ats", host=_HOST, port=_PORT) log.warning("auth DISABLED — no ~/.secrets/ats.env and no wallet reachable; loopback only") def _text(res): txt = "".join(c.get("text", "") for c in res.get("content", [])) if res.get("isError"): raise RuntimeError(txt) return txt def _run(leaf, tool, **kw): return _text(leaf.call_tool(tool, {k: v for k, v in kw.items() if k != "intent"})) def _crankd_crank(component, intent): """Forward a crank to crankd, which verifies the ticket's wallet approval and runs it OFF the agent's process tree. The agent no longer spawns a crank — it makes this one REST call.""" body = json.dumps({"component": component, "intent": intent}).encode() req = urllib.request.Request(CRANKD + "/crank", data=body, method="POST", headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=1200) as r: return r.read().decode() or "{}" except urllib.error.HTTPError as e: raise RuntimeError("crankd refused (HTTP %s): %s" % (e.code, e.read().decode())) except OSError as e: raise RuntimeError("crankd unreachable at %s: %s" % (CRANKD, e)) def _imgd(method, path, obj=None): """Forward an image op to imgd (the drivers run in imgd, not the gate). Returns the leaf's text.""" data = json.dumps(obj).encode() if obj is not None else None req = urllib.request.Request(IMGD + path, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=1800) as r: return json.loads(r.read().decode() or "{}").get("text", "") except urllib.error.HTTPError as e: raise RuntimeError("imgd refused (HTTP %s): %s" % (e.code, e.read().decode())) except OSError as e: raise RuntimeError("imgd unreachable at %s: %s" % (IMGD, e)) def _ws_read(path): """Read-only, path-confined read/list within the workstation repo — so the cranker can SEE the code and conventions before authoring (instead of guessing blind). No writes, no .git, no escaping.""" full = os.path.realpath(os.path.join(WORKSTATION, path or ".")) if not (full == WORKSTATION or full.startswith(WORKSTATION + os.sep)): raise RuntimeError("read refused: %r escapes the workstation repo" % path) if ".git" in full.split(os.sep): raise RuntimeError("read refused: .git is off-limits") if os.path.isdir(full): return ("dir %s/\n" % (path or ".")) + "\n".join(" " + e for e in sorted(os.listdir(full))) if not os.path.isfile(full): raise RuntimeError("no such path: %s" % path) return open(full, "r", errors="replace").read()[:20000] def _cong_read(path): """Read-only, path-confined read/list within the congruency repo — so the congruency cranker can SEE existing migrations/invocators/tools + conventions before authoring. No writes, no .git, no escaping.""" full = os.path.realpath(os.path.join(CONGRUENCY_REPO, path or ".")) if not (full == CONGRUENCY_REPO or full.startswith(CONGRUENCY_REPO + os.sep)): raise RuntimeError("read refused: %r escapes the congruency repo" % path) if ".git" in full.split(os.sep): raise RuntimeError("read refused: .git is off-limits") if os.path.isdir(full): return ("dir %s/\n" % (path or ".")) + "\n".join(" " + e for e in sorted(os.listdir(full))) if not os.path.isfile(full): raise RuntimeError("no such path: %s" % path) return open(full, "r", errors="replace").read()[:20000] # crankws (status -> ticket -> crank) — the workstation cranker @mcp.tool() def crankws_status(intent: str = "") -> str: return _run(crankws, "status") @mcp.tool() def crankws_read(path: str = ".", intent: str = "") -> str: """Read a file (or list a dir) in the workstation repo — read-only, confined to the repo. Use it to see existing modules + conventions before authoring, so you don't crank blind.""" return _ws_read(path) def _record_mandate(ticket_id, mandate): """Record each non-empty line of `mandate` as an ordered, content-addressed clause of the ticket's mandate (the ticket-class contract the change must satisfy). Returns the count filed.""" n = 0 for clause in (c.strip() for c in (mandate or "").splitlines()): if not clause: continue n += 1 dig = hashlib.sha256(("%s|%d|%s" % (ticket_id, n, clause)).encode()).hexdigest() _cong("POST", "/?route=mandate.record", {"ticket_id": int(ticket_id), "seq": n, "clause": clause, "digest": dig}) return n def _record_bindings(ticket_id, bindings): """Record each `<clause_seq> <guarantee label>` line as a declared binding — the author's claim that a shadow assertion satisfies a mandate clause. crankd applies these at harvest, setting the harvested guarantee's clause_seq. Returns the count filed.""" n = 0 for line in (l.strip() for l in (bindings or "").splitlines()): if not line: continue parts = line.split(None, 1) if len(parts) != 2 or not parts[0].isdigit(): continue cs, label = int(parts[0]), parts[1].strip() dig = hashlib.sha256(("%s|%d|%s" % (ticket_id, cs, label)).encode()).hexdigest() _cong("POST", "/?route=binding.record", {"ticket_id": int(ticket_id), "clause_seq": cs, "guarantee_label": label, "digest": dig}) n += 1 return n @mcp.tool() def crankws_ticket(title: str, body: str = "", patch: str = "", severity: str = "info", mandate: str = "", bindings: str = "", intent: str = "") -> str: """File a workstation crank ticket carrying its change (a confined python patch), its mandate (newline-separated falsifiable clauses the change must satisfy), and optionally its bindings — newline `<clause#> <shadow ✓ label>` lines declaring which shadow assertion satisfies which clause, which crankd applies when it auto-harvests guarantees at crank time (so coverage computes itself). The gate holds the congruency key; the agent never sees it. crankd applies the patch under the audit-veto at crank time.""" r = _cong("POST", "/?route=ticket.create", {"component": "workstation", "title": title, "body": body, "patch": patch, "severity": severity}) tid = r.get("rowid") if isinstance(r, dict) else None clauses = _record_mandate(tid, mandate) if tid else 0 binds = _record_bindings(tid, bindings) if tid else 0 return "filed workstation ticket %s (%d mandate clauses, %d bindings): %s" % (tid, clauses, binds, json.dumps(r)) @mcp.tool() def crankws_crank(message: str = "", intent: str = "") -> str: return _crankd_crank("workstation", intent or message) # crankcong (status -> read -> ticket -> crank) — the scoped congruency cranker (mirrors crankws) @mcp.tool() def crankcong_read(path: str = ".", intent: str = "") -> str: """Read a file (or list a dir) in the congruency repo — read-only, confined to the repo. See the migrations/, invocators/tags/dev/, tools/, setup.py conventions before authoring so you don't crank blind.""" return _cong_read(path) @mcp.tool() def crankcong_status(intent: str = "") -> str: """The congruency crank status: current version + the latest congruency ticket.""" try: ver = json.load(open(os.path.join(CONGRUENCY_REPO, "install.json"))).get("version", "?") except Exception: # noqa: BLE001 ver = "?" rows = (_cong("GET", "/?route=ticket.approval&component=congruency") or {}).get("rows") or [] t = rows[0] if rows else {} return "congruency version=%s · latest ticket #%s (%s) %s" % (ver, t.get("id"), t.get("status"), (t.get("title") or "")[:60]) @mcp.tool() def crankcong_ticket(title: str, body: str = "", patch: str = "", severity: str = "info", mandate: str = "", bindings: str = "", intent: str = "") -> str: """File a CONGRUENCY crank ticket carrying its change (a confined python patch), its mandate (newline falsifiable clauses the change must satisfy), and optional bindings (newline `<clause#> <shadow ✓ label>`). crankd applies the patch under the audit-veto and mints the next congruency version at crank time. The gate holds the congruency key; the agent never sees it.""" r = _cong("POST", "/?route=ticket.create", {"component": "congruency", "title": title, "body": body, "patch": patch, "severity": severity}) tid = r.get("rowid") if isinstance(r, dict) else None clauses = _record_mandate(tid, mandate) if tid else 0 binds = _record_bindings(tid, bindings) if tid else 0 return "filed congruency ticket %s (%d mandate clauses, %d bindings): %s" % (tid, clauses, binds, json.dumps(r)) @mcp.tool() def crankcong_crank(message: str = "", intent: str = "") -> str: return _crankd_crank("congruency", intent or message) # crankimg (pi image) — thin REST clients to imgd; the drivers run in imgd, not the gate @mcp.tool() def crankimg_build(node: str = "pi-forge", ssid: str = "WIFI-B61C", wifi_secret: str = "WIFI-B61C", intent: str = "") -> str: return _imgd("POST", "/build", {"node": node, "ssid": ssid, "wifi_secret": wifi_secret}) @mcp.tool() def crankimg_verify(intent: str = "") -> str: return _imgd("GET", "/verify") @mcp.tool() def crankimg_disks(intent: str = "") -> str: return _imgd("GET", "/disks") @mcp.tool() def crankimg_flash(device: str, confirm: str, intent: str = "") -> str: return _imgd("POST", "/flash", {"device": device, "confirm": confirm}) @mcp.tool() def crankimg_base(fetch: bool = False, intent: str = "") -> str: return _imgd("POST", "/base", {"fetch": fetch}) if __name__ == "__main__": transport = os.environ.get("ATS_TRANSPORT", "streamable-http") log.info("start transport=%s host=%s port=%s log=%s", transport, mcp.settings.host, mcp.settings.port, LOG_PATH) sys.stderr.write("[ats] FastMCP %s on %s:%s — intent + ticket + provenance enforced server-side; log=%s\n" % (transport, mcp.settings.host, mcp.settings.port, LOG_PATH)) mcp.run(transport=transport) ===END SOURCE=== THE CALLABLE SURFACE — every callable in the original, lowered with codebean's `lower_node` (#183). The rewrite must present THIS surface (same names, same parameters, same defaults) so callers of the original keep working: _secret(name) _setup_logging() _leaf(rel, name) now_iso() _actor() _cong(method, path, obj=None) record(tool, intent, note, ts) open_ticket() _tool_scopes(name) current_scopes() enforce(name, arguments) async GatedMCP.call_tool(self, name, arguments) async GatedMCP.list_tools(self) WalletTokenVerifier.__init__(self, token_map) async WalletTokenVerifier.verify_token(self, token) _text(res) _run(leaf, tool, **kw) _crankd_crank(component, intent) _imgd(method, path, obj=None) _ws_read(path) _cong_read(path) crankws_status(intent: str='') -> str crankws_read(path: str='.', intent: str='') -> str _record_mandate(ticket_id, mandate) _record_bindings(ticket_id, bindings) crankws_ticket(title: str, body: str='', patch: str='', severity: str='info', mandate: str='', bindings: str='', intent: str='') -> str crankws_crank(message: str='', intent: str='') -> str crankcong_read(path: str='.', intent: str='') -> str crankcong_status(intent: str='') -> str crankcong_ticket(title: str, body: str='', patch: str='', severity: str='info', mandate: str='', bindings: str='', intent: str='') -> str crankcong_crank(message: str='', intent: str='') -> str crankimg_build(node: str='pi-forge', ssid: str='WIFI-B61C', wifi_secret: str='WIFI-B61C', intent: str='') -> str crankimg_verify(intent: str='') -> str crankimg_disks(intent: str='') -> str crankimg_flash(device: str, confirm: str, intent: str='') -> str crankimg_base(fetch: bool=False, intent: str='') -> str CAPTURED BEHAVIORAL EXEMPLARS — real (input, output) pairs mined from THIS module with criteria_miner (#200's "verified, not asserted"), running the ORIGINAL as the oracle. THE REWRITE MUST REPRODUCE EVERY PAIR BELOW, exactly: _record_mandate(ticket_id, mandate) — 19 exemplar(s) captured: _record_mandate(-1, 0) == 0 _record_mandate(-1, '') == 0 _record_mandate(-1, []) == 0 _record_mandate(0, 0) == 0 _record_mandate(0, '') == 0 _record_mandate(0, []) == 0 ... 13 more exemplar(s) for this function — re-mine them with criteria_miner.mine() on the source above _record_bindings(ticket_id, bindings) — 30 exemplar(s) captured: _record_bindings(-1, 0) == 0 _record_bindings(-1, 'POST') == 0 _record_bindings(-1, '%s|%d|%s') == 0 _record_bindings(-1, 'guarantee_label') == 0 _record_bindings(-1, '/?route=binding.record') == 0 _record_bindings(-1, 'clause_seq') == 0 ... 24 more exemplar(s) for this function — re-mine them with criteria_miner.mine() on the source above HONEST LIMIT (criteria_miner's own): these cover the SAMPLED inputs, probed around the function's own branch constants. They are a strong refactor-drift net, NOT a proof of total equivalence — so treat them as the FLOOR of this ticket's equivalence evidence and add the edge cases they miss (see ACCEPTANCE). THE STANDARD — #200's four requirements, which this ticket exists to enforce. All four are the standard's own words; none is optional: 1. BEHAVIOR-PRESERVING, VERIFIED NOT ASSERTED — the deliverable is ENTIRELY NEW source code (rewritten, not copied). It must reproduce the ORIGINAL's behavior, and the proof is the exemplars above (captured via codebean lower_* #183 + criteria_miner IO-exemplars, exactly as #200 specifies). 2. A PYTHON MODULE — regardless of the source language. This module's source is `.py`; the rewrite is Python either way. 3. INSTALLABLE — a proper package: pyproject.toml, `pip install -e .` works, console entry points where applicable. WRITE ONLY YOUR OWN MODULE FILE at forge2/<repo>/<module>.py. Do NOT hand-author pyproject.toml or __init__.py: the harness DERIVES them ADDITIVELY from the modules present (congruency/tools/ forge_package.py, #410 — the analogue of the tag-registry fix #281), so your unit never conflicts with, or clobbers, a sibling module of the same forge2/<repo>/ package. A package-level pyproject includes every .py in the directory, so your module installs and imports as `<repo>.<module>` the moment it lands. 4. CONTROL-CONFIG'D — NO hardcoded paths or params. A JSON/registry config object drives it (the install.json / registry.json idiom of this repo), and per note-for-claude a tool that cannot see its config THROWS rather than guessing a default. Add ONLY your own `forge2.<repo>.<module>` section to the package's registry.json (create the file if it is absent); NEVER read, edit or drop a sibling module's section — the sections are disjoint keys and stay additive that way. ACCEPTANCE (#200's, per rewrite ticket) — this ticket settles when ALL of: a. the new source reproduces every captured exemplar above, plus the edge cases you add for what sampling cannot reach (paste the runs as evidence); b. it presents the callable surface above, so existing callers keep working; c. `pip install -e .` installs it clean from its own pyproject.toml; d. it runs off a config object with ZERO hardcoded settings, and throws without one. Kind is `build` (a pair: coder + adversarial tester). This ticket hangs off #200 by a `spawned` edge, so a stop ticket aimed at #200 brakes every rewrite in this programme. Filed mechanically by checkouts/current/congruency/tools/forge_refactor_ticketer.py under ticket #201, from the forge bare repos at /Users/stevenpeterson/code/installed/jazz-project/config/forge/remotes. Its original source is embedded above (#346), so this ticket is complete on its own.

— browse congruency source (forge-parked tree)