port
Every small JSONL tool declares its own port-verb — source, transform, filter, fold, or sink — in a manifest field or a --port flag, so a map or a typechecker can READ a tool's composition shape instead of guessing it. The load-bearing move is `port check`: when a tool declares its verb in both places, they must agree, and a drift is a decidable non-zero exit, not a thing a human notices later.
The honest edge
It verifies a declaration is consistent with itself — manifest field vs the tool's own --port flag — not that the declared verb is true of the tool's actual behavior. A tool can honestly declare 'filter' in both places and still behave like a transform in its code; proving a verb against real behavior is a deeper, undecidable-in-general question this tool does not claim to answer.
Run it
python3 port.py verbs
test_port.py (15/15, mutation-bitten)
Python 3, standard library only
The code — every file that ships
port.py239 lineson GitHub →
#!/usr/bin/env python3
"""port — declare a composition tool's port-verb, and read it back honestly.
A composition of small JSONL tools only typechecks if each tool declares its
own PORT-VERB: the type-level shape of its stdin/stdout contract on the shared
JSON-lines interface. Five verbs, and only five:
source : nothing -> JSONL (emits; no meaningful stdin)
transform : JSONL -> JSONL (record in, record out)
filter : JSONL -> JSONL' (record in, subset out — output type <= input)
fold : JSONL -> JSONL_agg (records in, an aggregate/narrower record out)
sink : JSONL -> nothing (terminal side effect, no pipeable stdout)
WHY THIS TOOL EXISTS. Before `port`, a map or a typechecker that wanted to know
a tool's port-verb had to GUESS it — from a one-line description, or from a
hand-kept list living in one program's source. A guess that a tool emits when it
actually sinks is a silent lie: the map draws an edge that cannot carry data.
`port` removes the guess. Every tool DECLARES its port-verb, in one of two honest
places, and `port` reads the declaration back:
1. A manifest field `port_verb` in a JSON manifest of tool entries, OR
2. A `--port` flag the tool answers for itself.
The load-bearing move is the CHECK (`port check`): when a tool declares its
port-verb in BOTH places, they must AGREE. A manifest that says `filter` while
the tool's own `--port` says `source` is a declaration that has drifted from the
thing it describes — and `port check` makes that drift a decidable, non-zero
exit, not a thing a human notices later. That is the whole point: a declaration
you can verify beats a description you have to trust.
port's own port-verb is `source`: it emits port declarations as JSON-lines and
takes no meaningful stdin.
No dependencies beyond the Python standard library. MIT licensed.
USAGE
port verbs
Print the five port-verbs and their type contracts (JSONL, one per line).
port read --manifest FILE [--slug SLUG]
Read declared port_verb(s) from a manifest. With --slug, one entry;
without, every entry that declares a port_verb. Emits JSON-lines:
{"slug": ..., "port_verb": ..., "source": "manifest"}
A manifest entry with no port_verb field is reported with
port_verb=null and status="undeclared" (flagged, never guessed).
port check --manifest FILE --flag-cmd 'CMD {slug}'
For every manifest entry that declares a port_verb, run the tool's own
--port flag (via --flag-cmd, with {slug} substituted) and compare.
Emits one JSON-line verdict per entry:
{"slug":..., "manifest":..., "flag":..., "status":"agree|drift|missing"}
Exit 0 iff every checked entry AGREES; exit 3 on any drift/missing.
This is the decidable self-check — a declaration verified against itself.
port emit --slug SLUG --port-verb VERB
Emit a single declaration JSON-line (for a tool that has no manifest yet).
"""
import argparse
import json
import subprocess
import sys
# The five port-verbs and their type-level contract on the shared JSONL interface.
# This dict is the ONLY definition of the vocabulary — the closed set.
PORT_VERBS = {
"source": "nothing -> JSONL (emits; no meaningful stdin)",
"transform": "JSONL -> JSONL (record in, record out)",
"filter": "JSONL -> JSONL' (record in, subset out; output type <= input)",
"fold": "JSONL -> JSONL_agg (records in, an aggregate/narrower record out)",
"sink": "JSONL -> nothing (terminal side effect, no pipeable stdout)",
}
def is_valid_verb(verb):
"""A port-verb is valid iff it is one of the five closed-set verbs."""
return verb in PORT_VERBS
def load_manifest_entries(path):
"""Return the list of tool entries from a manifest.
Accepts either a bare JSON list of entries, or a JSON object with a top-level
key ('gifts', 'tools', or 'entries') holding the list. Raises ValueError if
no entry list can be found — never guesses a shape.
"""
with open(path) as f:
data = json.load(f)
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ("gifts", "tools", "entries"):
if isinstance(data.get(key), list):
return data[key]
raise ValueError(
"manifest is neither a list of entries nor an object with a "
"'gifts'/'tools'/'entries' list"
)
def cmd_verbs(_args):
"""Emit the five port-verbs and their contracts, one JSON-line each."""
for verb, contract in PORT_VERBS.items():
print(json.dumps({"port_verb": verb, "contract": contract}))
return 0
def cmd_read(args):
"""Read declared port_verb(s) from a manifest and emit them as JSON-lines."""
entries = load_manifest_entries(args.manifest)
found = False
for entry in entries:
slug = entry.get("slug")
if args.slug is not None and slug != args.slug:
continue
verb = entry.get("port_verb")
if verb is None:
rec = {"slug": slug, "port_verb": None,
"source": "manifest", "status": "undeclared"}
elif not is_valid_verb(verb):
rec = {"slug": slug, "port_verb": verb,
"source": "manifest", "status": "invalid"}
else:
rec = {"slug": slug, "port_verb": verb,
"source": "manifest", "status": "declared"}
print(json.dumps(rec))
found = True
if args.slug is not None and not found:
print(json.dumps({"slug": args.slug, "port_verb": None,
"source": "manifest", "status": "not-found"}))
return 3
return 0
def _run_flag(flag_cmd, slug):
"""Run a tool's own --port flag and return the verb it prints, or None.
flag_cmd is a template string with '{slug}' substituted for the entry slug.
The tool is expected to print its port-verb as the first whitespace token
on stdout. Any failure (non-zero exit, no output, unparseable) returns None
— an honest 'the tool did not answer', never a guessed verb.
"""
cmd = flag_cmd.replace("{slug}", slug)
try:
out = subprocess.run(cmd, shell=True, capture_output=True,
text=True, timeout=30)
except Exception:
return None
if out.returncode != 0:
return None
token = out.stdout.strip().split()
if not token:
return None
verb = token[0]
return verb if is_valid_verb(verb) else None
def cmd_check(args):
"""Compare each manifest port_verb against the tool's own --port flag.
Exit 0 iff every checked entry agrees; exit 3 on any drift or missing.
"""
entries = load_manifest_entries(args.manifest)
all_agree = True
checked_any = False
for entry in entries:
slug = entry.get("slug")
manifest_verb = entry.get("port_verb")
if manifest_verb is None:
continue # nothing declared in the manifest -> nothing to check here
checked_any = True
flag_verb = _run_flag(args.flag_cmd, slug)
if flag_verb is None:
status = "missing"
all_agree = False
elif flag_verb == manifest_verb:
status = "agree"
else:
status = "drift"
all_agree = False
print(json.dumps({"slug": slug, "manifest": manifest_verb,
"flag": flag_verb, "status": status}))
if not checked_any:
# Nothing declared a port_verb: there is nothing to verify. This is not
# a pass (there was no check) — report it and exit 3 so a manifest that
# forgot every declaration cannot read as green.
print(json.dumps({"status": "no-declarations",
"note": "no manifest entry declared a port_verb"}))
return 3
return 0 if all_agree else 3
def cmd_emit(args):
"""Emit a single port declaration JSON-line for a manifest-less tool."""
if not is_valid_verb(args.port_verb):
print(json.dumps({"slug": args.slug, "port_verb": args.port_verb,
"status": "invalid",
"valid": list(PORT_VERBS)}))
return 3
print(json.dumps({"slug": args.slug, "port_verb": args.port_verb,
"source": "declared", "status": "declared"}))
return 0
def build_parser():
p = argparse.ArgumentParser(
prog="port",
description="Declare a composition tool's port-verb, and read it back honestly.",
)
sub = p.add_subparsers(dest="cmd", required=True)
sp = sub.add_parser("verbs", help="print the five port-verbs and contracts")
sp.set_defaults(func=cmd_verbs)
sp = sub.add_parser("read", help="read declared port_verb(s) from a manifest")
sp.add_argument("--manifest", required=True)
sp.add_argument("--slug", default=None)
sp.set_defaults(func=cmd_read)
sp = sub.add_parser("check", help="verify manifest port_verb against each tool's --port flag")
sp.add_argument("--manifest", required=True)
sp.add_argument("--flag-cmd", required=True,
help="command template to run a tool's --port flag; {slug} is substituted")
sp.set_defaults(func=cmd_check)
sp = sub.add_parser("emit", help="emit a single port declaration line")
sp.add_argument("--slug", required=True)
sp.add_argument("--port-verb", required=True)
sp.set_defaults(func=cmd_emit)
return p
def main(argv=None):
args = build_parser().parse_args(argv)
return args.func(args)
if __name__ == "__main__":
sys.exit(main())
test_port.py168 lineson GitHub →
#!/usr/bin/env python3
"""Mutation-bitten tests for the port gift.
Each test asserts a real behavior. To prove the suite is non-vacuous, run it
against a deliberately-broken port.py and confirm it goes RED (see the
`_self_mutation_note` at the bottom — the mutations are documented, not shipped).
"""
import json
import os
import subprocess
import sys
import tempfile
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
PORT = os.path.join(HERE, "port.py")
sys.path.insert(0, HERE)
import port # noqa: E402
def run(*args, input_text=None):
"""Run port.py as a subprocess; return (exit, stdout_lines)."""
proc = subprocess.run(
[sys.executable, PORT, *args],
capture_output=True, text=True, input=input_text,
)
lines = [json.loads(l) for l in proc.stdout.splitlines() if l.strip()]
return proc.returncode, lines
def write_manifest(entries, wrap=None):
"""Write a temp manifest; wrap=None -> bare list, else {wrap: [...]}."""
fd, path = tempfile.mkstemp(suffix=".json")
with os.fdopen(fd, "w") as f:
json.dump(entries if wrap is None else {wrap: entries}, f)
return path
class TestVocabulary(unittest.TestCase):
def test_exactly_five_verbs(self):
# The closed set is exactly five — no more, no fewer.
self.assertEqual(len(port.PORT_VERBS), 5)
self.assertEqual(set(port.PORT_VERBS),
{"source", "transform", "filter", "fold", "sink"})
def test_valid_verb_gate(self):
self.assertTrue(port.is_valid_verb("filter"))
self.assertFalse(port.is_valid_verb("emit")) # a plausible non-verb
self.assertFalse(port.is_valid_verb(""))
self.assertFalse(port.is_valid_verb(None))
def test_verbs_command_emits_five_lines(self):
code, lines = run("verbs")
self.assertEqual(code, 0)
self.assertEqual(len(lines), 5)
self.assertEqual({l["port_verb"] for l in lines}, set(port.PORT_VERBS))
class TestRead(unittest.TestCase):
def test_read_declared(self):
m = write_manifest([{"slug": "a", "port_verb": "source"},
{"slug": "b", "port_verb": "sink"}])
code, lines = run("read", "--manifest", m)
self.assertEqual(code, 0)
self.assertEqual(lines[0], {"slug": "a", "port_verb": "source",
"source": "manifest", "status": "declared"})
self.assertEqual(lines[1]["status"], "declared")
def test_read_undeclared_is_flagged_not_guessed(self):
# A gift with NO port_verb must come back status=undeclared, verb null —
# never a guessed verb. This is the whole reason the gift exists.
m = write_manifest([{"slug": "x", "verb": "survive"}]) # 'verb' is the
code, lines = run("read", "--manifest", m) # marketing verb
self.assertEqual(code, 0)
self.assertEqual(lines[0]["port_verb"], None)
self.assertEqual(lines[0]["status"], "undeclared")
def test_read_invalid_verb_flagged(self):
m = write_manifest([{"slug": "x", "port_verb": "emit"}]) # not a verb
code, lines = run("read", "--manifest", m)
self.assertEqual(lines[0]["status"], "invalid")
def test_read_slug_filter(self):
m = write_manifest([{"slug": "a", "port_verb": "source"},
{"slug": "b", "port_verb": "sink"}])
code, lines = run("read", "--manifest", m, "--slug", "b")
self.assertEqual(len(lines), 1)
self.assertEqual(lines[0]["slug"], "b")
def test_read_slug_not_found_exits_3(self):
m = write_manifest([{"slug": "a", "port_verb": "source"}])
code, lines = run("read", "--manifest", m, "--slug", "zzz")
self.assertEqual(code, 3)
self.assertEqual(lines[0]["status"], "not-found")
def test_read_wrapped_manifest(self):
# Real gifts-manifest.json wraps the list under 'gifts'.
m = write_manifest([{"slug": "a", "port_verb": "fold"}], wrap="gifts")
code, lines = run("read", "--manifest", m)
self.assertEqual(code, 0)
self.assertEqual(lines[0]["port_verb"], "fold")
class TestCheck(unittest.TestCase):
def _flag_cmd(self, mapping):
"""Build a --flag-cmd that echoes a per-slug verb via a tiny py snippet."""
table = json.dumps(mapping)
# {slug} is substituted by port; the snippet prints the mapped verb.
return (f"{sys.executable} -c "
f"'import json,sys; "
f"print(json.loads(sys.argv[1]).get(sys.argv[2],\"\"))' "
f"'{table}' {{slug}}")
def test_check_agree_exits_0(self):
m = write_manifest([{"slug": "a", "port_verb": "source"}])
code, lines = run("check", "--manifest", m,
"--flag-cmd", self._flag_cmd({"a": "source"}))
self.assertEqual(code, 0)
self.assertEqual(lines[0]["status"], "agree")
def test_check_drift_exits_3(self):
# Manifest says source, the tool's own flag says sink -> drift, exit 3.
m = write_manifest([{"slug": "a", "port_verb": "source"}])
code, lines = run("check", "--manifest", m,
"--flag-cmd", self._flag_cmd({"a": "sink"}))
self.assertEqual(code, 3)
self.assertEqual(lines[0]["status"], "drift")
def test_check_missing_flag_exits_3(self):
# Tool answers nothing -> missing, not silently passed.
m = write_manifest([{"slug": "a", "port_verb": "source"}])
code, lines = run("check", "--manifest", m,
"--flag-cmd", self._flag_cmd({})) # empty -> ""
self.assertEqual(code, 3)
self.assertEqual(lines[0]["status"], "missing")
def test_check_no_declarations_is_not_green(self):
# A manifest that declared no port_verb at all must NOT read as pass.
m = write_manifest([{"slug": "a", "verb": "survive"}])
code, lines = run("check", "--manifest", m,
"--flag-cmd", self._flag_cmd({"a": "source"}))
self.assertEqual(code, 3)
self.assertEqual(lines[0]["status"], "no-declarations")
class TestEmit(unittest.TestCase):
def test_emit_valid(self):
code, lines = run("emit", "--slug", "port", "--port-verb", "source")
self.assertEqual(code, 0)
self.assertEqual(lines[0]["port_verb"], "source")
self.assertEqual(lines[0]["status"], "declared")
def test_emit_invalid_exits_3(self):
code, lines = run("emit", "--slug", "x", "--port-verb", "nonsense")
self.assertEqual(code, 3)
self.assertEqual(lines[0]["status"], "invalid")
if __name__ == "__main__":
unittest.main(verbosity=2)
# _self_mutation_note: this suite was proven non-vacuous by mutation —
# (1) PORT_VERBS drop 'sink' -> test_exactly_five_verbs RED
# (2) cmd_read guess 'source' on null -> test_read_undeclared_* RED
# (3) cmd_check return 0 always -> test_check_drift/missing RED
# (4) cmd_check treat no-decl as pass -> test_check_no_declarations RED
# each mutation reverted after confirming RED. A no-op runner cannot pass these.