"""Phase 4b — layer demo: tenant fork -> private extension -> sanitized PR. Story shown in Gitea: 1. tenant-acme forks skills-community/ (private tenant copy) 2. commit A (tenant fork only): extensions/acme/hiring-process.md with ACME-internal specifics (names, mail, intranet URL) — stays private 3. commit B on branch 'publish-hiring-process': the GENERALIZED, anonymized variant extensions/community/structured-hiring-process.md 4. pull request against skills-community/ with a before/after diff in the description; sanitize.yml + eval.yml run as PR gates Idempotent: existing fork/branch/PR are detected and reused/updated. """ import os import shutil import subprocess import sys import tempfile import requests from dotenv import load_dotenv BASE = os.path.join(os.path.dirname(__file__), "..") load_dotenv(os.path.join(BASE, ".env")) GITEA_URL = os.environ["GITEA_URL"].rstrip("/") TOKEN = os.environ["GITEA_TOKEN"] API = f"{GITEA_URL}/api/v1" HDR = {"Authorization": f"token {TOKEN}"} SKILLS_DIR = os.path.join(BASE, "skills") BRANCH = "publish-hiring-process" # Commit A — tenant-internal (deliberately contains PII-like content; it only # ever exists in the PRIVATE tenant repo and is NOT part of the PR). ACME_INTERNAL = """# ACME hiring process (internal) Owner: Jane Miller — HR office, ext. +1 555 0187 Process wiki: http://intranet.acme.local/hr/hiring ## Our process 1. Screening call by the recruiting team (30 min, template on the intranet). 2. Two structured interviews (hiring manager + peer). 3. Case study: role-specific take-home task, reviewed by the team lead. 4. Offer within 5 working days; tooling: Personio (ATS), contract via DocuSign. """ # Commit B — generalized, anonymized (this is what the PR proposes). COMMUNITY_VARIANT = """# Structured hiring process (community pattern) A field-tested interview loop contributed by a tenant (anonymized). ## Pattern 1. Screening call by the recruiting team (30 minutes, standardized template). 2. Two structured interviews: one with the hiring manager, one with a peer. 3. Role-specific case study (take-home), reviewed by the future team lead. 4. Offer decision within five working days. ## Why it works - Structured interviews reduce bias and improve signal quality. - The case study validates practical skills instead of interview performance. - A fixed decision deadline keeps the candidate experience predictable. Tooling category: applicant tracking system (ATS) + e-signature solution. """ PR_BODY = """## Tenant contribution: structured hiring process This PR proposes a **generalized, anonymized** version of a hiring process that proved successful at a tenant. The tenant-internal original stays in the private tenant repo. ### Before (tenant-internal, NOT part of this PR) ```markdown Owner: J*** M*** - HR office, ext. +1 *** **** Process wiki: http://intranet.acme.local/hr/hiring ... 4. Offer within 5 working days; tooling: Personio (ATS), contract via DocuSign. ``` ### After (this PR) ```markdown 1. Screening call by the recruiting team (30 minutes, standardized template). 2. Two structured interviews: one with the hiring manager, one with a peer. 3. Role-specific case study (take-home), reviewed by the future team lead. 4. Offer decision within five working days. Tooling category: applicant tracking system (ATS) + e-signature solution. ``` **Gate checklist** - [x] personal data removed (names, e-mail, phone) - [x] internal URLs removed - [x] vendor names generalized to tool categories - [x] structural eval passes (`eval.yml`) - [x] PII scan passes (`sanitize.yml`) """ def api(method, path, ok=(200, 201, 204), **kw): r = requests.request(method, f"{API}{path}", headers=HDR, timeout=120, **kw) if r.status_code not in ok: raise RuntimeError(f"{method} {path} -> {r.status_code}: {r.text[:300]}") return r def run_git(cwd, *args): return subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True, text=True) def main(): recruiter = next((d for d in sorted(os.listdir(SKILLS_DIR)) if "recruit" in d), None) if not recruiter: sys.exit("no recruiter package — run p2 first") # 1) fork into tenant-acme (idempotent) r = requests.get(f"{API}/repos/tenant-acme/{recruiter}", headers=HDR, timeout=60) if r.status_code != 200: api("POST", f"/repos/skills-community/{recruiter}/forks", json={"organization": "tenant-acme"}, ok=(200, 201, 202)) print(f"forked skills-community/{recruiter} -> tenant-acme") auth_url = GITEA_URL.replace("://", f"://gitadmin:{TOKEN}@") tmp = tempfile.mkdtemp(prefix="sftenant_") try: run_git(".", "clone", f"{auth_url}/tenant-acme/{recruiter}.git", tmp) run_git(tmp, "config", "user.name", "acme-hr") run_git(tmp, "config", "user.email", "hr@acme-corp.example") # 2) commit A on main: tenant-internal extension (stays private) acme_dir = os.path.join(tmp, "extensions", "acme") os.makedirs(acme_dir, exist_ok=True) with open(os.path.join(acme_dir, "hiring-process.md"), "w", encoding="utf-8", newline="\n") as f: f.write(ACME_INTERNAL) run_git(tmp, "add", "-A") if subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=tmp).returncode != 0: run_git(tmp, "commit", "-m", "feat(acme): internal hiring process (tenant layer)") run_git(tmp, "push", "origin", "main") print("tenant-internal extension committed (private)") # 3) commit B on publish branch: generalized variant only run_git(tmp, "checkout", "-B", BRANCH, "origin/main") # the publish branch must NOT carry the internal file internal = os.path.join(tmp, "extensions", "acme", "hiring-process.md") if os.path.exists(internal): run_git(tmp, "rm", "-q", "extensions/acme/hiring-process.md") comm_dir = os.path.join(tmp, "extensions", "community") os.makedirs(comm_dir, exist_ok=True) with open(os.path.join(comm_dir, "structured-hiring-process.md"), "w", encoding="utf-8", newline="\n") as f: f.write(COMMUNITY_VARIANT) run_git(tmp, "add", "-A") if subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=tmp).returncode != 0: run_git(tmp, "commit", "-m", "feat(community): generalized structured hiring process (sanitized)") run_git(tmp, "push", "-f", "origin", BRANCH) print(f"publish branch pushed: tenant-acme/{recruiter}:{BRANCH}") finally: shutil.rmtree(tmp, ignore_errors=True) # 4) PR against skills-community (idempotent: skip if open PR exists) prs = api("GET", f"/repos/skills-community/{recruiter}/pulls?state=open").json() if not any(p["head"]["label"].endswith(BRANCH) for p in prs): pr = api("POST", f"/repos/skills-community/{recruiter}/pulls", json={ "title": "Tenant contribution: structured hiring process (sanitized)", "body": PR_BODY, "head": f"tenant-acme:{BRANCH}", "base": "main", }).json() print(f"PR created: {pr['html_url']}") else: print("PR already open — skipped") if __name__ == "__main__": main()