hololake-system-architecture/engineering/persona-history-runtime/runtime/guanghu_history_publisher.py

#!/usr/bin/env python3
"""Publish sanitized BS-SH-005 recovery watermarks to two allowlisted repos."""

from __future__ import annotations

import argparse
import json
import os
import pathlib
import subprocess
import urllib.request
from datetime import datetime, timezone


ALLOWED_TARGETS = {
    "REPO-014": (
        "https://guanghulab.com/code/bingshuo/hololake-system-architecture.git",
        "operations/BS-SH-005/persona-history/CURRENT.json",
    ),
    "REPO-012": (
        "https://guanghulab.com/code/bingshuo/guanghu-ice-heart.git",
        "eternal-lake-heart/heartbeat-core/zhuyuan-persona-system/"
        "server-watermarks/BS-SH-005-PERSONA-HISTORY-CURRENT.json",
    ),
}
FORBIDDEN_PUBLIC_FIELDS = {
    "private_locator",
    "email",
    "password",
    "secret",
    "token",
    "api_key",
    "private_key",
}


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def validate_config(config: dict) -> None:
    if config.get("node_id") != "BS-SH-005":
        raise ValueError("publisher node is not BS-SH-005")
    repositories = config.get("repositories", [])
    if {item.get("id") for item in repositories} != set(ALLOWED_TARGETS):
        raise ValueError("publisher repository allowlist mismatch")
    for item in repositories:
        if (item.get("url"), item.get("snapshot_path")) != ALLOWED_TARGETS[item["id"]]:
            raise ValueError(f"publisher target mismatch for {item['id']}")


def validate_public_snapshot(snapshot: dict) -> None:
    encoded = json.dumps(snapshot, ensure_ascii=False).lower()
    for field in FORBIDDEN_PUBLIC_FIELDS:
        if f'"{field.lower()}"' in encoded:
            raise ValueError(f"private field rejected: {field}")
    if snapshot.get("node_id") != "BS-SH-005":
        raise ValueError("snapshot node mismatch")
    if snapshot.get("persona_state") not in {"NOT_BORN", "BIRTH_GATE_PENDING"}:
        raise ValueError("unexpected persona state")


def material_snapshot(snapshot: dict) -> dict:
    """Drop heartbeat-only timestamps so idle cycles never create Git commits."""
    normalized = json.loads(json.dumps(snapshot))
    normalized.pop("updated_at", None)
    for source in normalized.get("sources", {}).values():
        source.pop("updated_at", None)
    return normalized


def git_env(config: dict) -> dict[str, str]:
    env = dict(os.environ)
    env.update(
        {
            "HOME": config["state_root"],
            "GIT_TERMINAL_PROMPT": "0",
            "GIT_CONFIG_COUNT": "1",
            "GIT_CONFIG_KEY_0": "credential.helper",
            "GIT_CONFIG_VALUE_0": config["credential_helper"],
        }
    )
    return env


def run_git(env: dict[str, str], *args: str, cwd: pathlib.Path | None = None) -> None:
    subprocess.run(
        ["git", *args],
        cwd=cwd,
        env=env,
        check=True,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        text=True,
        timeout=300,
    )


def atomic_json(path: pathlib.Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    pending = path.with_name(f".{path.name}.{os.getpid()}.pending")
    pending.write_text(
        json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(pending, path)


def publish_repository(
    config: dict, repository: dict, snapshot: dict, env: dict[str, str]
) -> None:
    checkout = pathlib.Path(config["state_root"]) / "repositories" / repository["id"]
    if not (checkout / ".git").is_dir():
        checkout.parent.mkdir(parents=True, exist_ok=True)
        run_git(
            env,
            "clone",
            "--branch",
            "main",
            "--single-branch",
            repository["url"],
            str(checkout),
        )
    else:
        run_git(env, "fetch", "origin", "main", cwd=checkout)
        run_git(env, "merge", "--ff-only", "origin/main", cwd=checkout)

    target = checkout / repository["snapshot_path"]
    published = material_snapshot(snapshot)
    if target.is_file():
        current = json.loads(target.read_text(encoding="utf-8"))
        current.pop("publisher", None)
        if current == published:
            return
    published["publisher"] = {
        "schema": "guanghu.persona-history-publication/v1",
        "node_id": "BS-SH-005",
        "repository_id": repository["id"],
        "published_at": now_iso(),
        "source_observed_at": snapshot.get("updated_at"),
        "content_policy": "SANITIZED_PUBLIC_WATERMARK_ONLY",
    }
    atomic_json(target, published)
    run_git(env, "add", "--", repository["snapshot_path"], cwd=checkout)
    changed = subprocess.run(
        ["git", "diff", "--cached", "--quiet", "--exit-code"],
        cwd=checkout,
        env=env,
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        text=True,
    ).returncode
    if changed == 0:
        return
    if changed != 1:
        raise RuntimeError("git staged diff check failed")
    run_git(
        env,
        "-c",
        "user.name=BS-SH-005 Persona History",
        "-c",
        "user.email=bs-sh-005@noreply.guanghulab.com",
        "commit",
        "-m",
        "chore(history): update BS-SH-005 recovery watermark",
        cwd=checkout,
    )
    run_git(env, "push", "origin", "HEAD:main", cwd=checkout)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True, type=pathlib.Path)
    args = parser.parse_args()
    config = json.loads(args.config.read_text(encoding="utf-8"))
    validate_config(config)
    with urllib.request.urlopen(config["snapshot_url"], timeout=15) as response:
        snapshot = json.loads(response.read())
    validate_public_snapshot(snapshot)
    env = git_env(config)
    for repository in config["repositories"]:
        publish_repository(config, repository, snapshot, env)


if __name__ == "__main__":
    main()