railiance-platform/scripts/fluid_telegram_custody_move.py

#!/usr/bin/env python3
"""RPF-WP-0035-T03: attended, silent move of fluid-telegram custody to tenant coulomb.

Founder ruling 2026-10-09 (ops-mason message b72ff77d): the hall-of-helix entries
found under the undocumented infotech prefix move to the approved coulomb prefix
that CCR-2026-0039 grants. For each of bot-token, operator-app, operator-session
and redaction-salt this process reads the current version at
``platform/workloads/infotech/fluid-telegram/hall-of-helix/<entry>``, writes it to
``platform/workloads/coulomb/fluid-telegram/hall-of-helix/<entry>`` with CAS 0
and verifies the readback. Only after all four are verified, and no source
changed during the run, are the infotech entries metadata-deleted. Values only
travel through this process's memory and the ``bao`` child's stdin.

Run from a railiance-platform checkout at the reviewed commit:

  BAO_ADDR=http://127.0.0.1:18200 VAULT_ADDR=http://127.0.0.1:18200 \\
  warden access openbao-platform-admin-login --exec -- \\
    python3 scripts/fluid_telegram_custody_move.py \\
      --confirm 'MOVE fluid-telegram hall-of-helix to coulomb' \\
      --receipt "$HOME/ft-custody-move-receipt.json"

``--dry-run`` touches only KV metadata. ``--keep-source`` copies and verifies
but does not delete. A rerun after a completed move reports ``already_moved``.

The process prints nothing (stdout/stderr go to /dev/null). The optional
receipt (mode 0600, refused if it exists) holds versions, booleans and an
outcome code only.

Exit codes:
  0  moved (or already moved / already identical); dry-run: a move would run
  1  unexpected internal error (contained)
  2  precondition refused before any write (confirmation, envelope, identity,
     an entry absent at both prefixes, absence not provable, receipt exists)
  3  a coulomb entry exists and differs, or a current version is not live;
     never overwritten
  4  write or delete failed
  5  verify failed: readback differs, or a source changed during the run
  6  dry-run only: some coulomb entries already exist
"""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parent))
from tenant_engine_audit_core_sender_seed import Exit, bao, live_version, read_json  # noqa: E402

SOURCE = 'workloads/infotech/fluid-telegram/hall-of-helix'
TARGET = 'workloads/coulomb/fluid-telegram/hall-of-helix'
ENTRIES = ('bot-token', 'operator-app', 'operator-session', 'redaction-salt')
CONFIRM = 'MOVE fluid-telegram hall-of-helix to coulomb'
SCHEMA = 'platform.fluid-telegram-custody-move.v1'


def meta(prefix, entry):
    return read_json(f'platform/metadata/{prefix}/{entry}', absent_ok=True)


def data(prefix, entry, code=2):
    raw = read_json(f'platform/data/{prefix}/{entry}', absent_ok=False, code=code)
    try:
        return raw['data'], raw['metadata']['version']
    except (KeyError, TypeError):
        raise Exit(code, 'unparseable_read') from None


def run(args, receipt):
    if args.confirm != CONFIRM:
        raise Exit(2, 'exact_confirmation_required')
    if (Path.home().parent.name != '.warden-attended-login'
            or os.environ.get('BAO_TOKEN') or os.environ.get('VAULT_TOKEN')):
        raise Exit(2, 'attended_warden_envelope_required')
    ident = read_json('auth/token/lookup-self', absent_ok=False)
    policies = ident.get('policies') or []
    if 'platform-admin' not in policies or 'root' in policies:
        raise Exit(2, 'attended_platform_admin_required')

    rows = receipt['entries']
    plan = []
    for entry in ENTRIES:
        src, dst = meta(SOURCE, entry), meta(TARGET, entry)
        row = rows[entry] = {'source_present': src is not None, 'target_present': dst is not None}
        if src is None and dst is None:
            raise Exit(2, 'entry_absent_at_both_prefixes')
        if src is not None:
            if live_version(src) is None:
                raise Exit(3, 'source_current_version_not_live')
            row['source_version'] = src['current_version']
        if dst is not None:
            if live_version(dst) is None:
                raise Exit(3, 'target_current_version_not_live')
            row['target_version'] = dst['current_version']
        plan.append((entry, src, dst))

    if args.dry_run:
        receipt['value_reads'] = 0
        if any(dst is not None for _, src, dst in plan if src is not None):
            raise Exit(6, 'dry_run_target_present')
        return 'dry_run_would_move' if any(src for _, src, _ in plan) else 'dry_run_already_moved'

    if all(src is None for _, src, _ in plan):
        return 'already_moved'

    # Phase 1: copy and verify every entry before anything is deleted.
    for entry, src, dst in plan:
        if src is None:
            continue
        value, version = data(SOURCE, entry)
        if version != src['current_version']:
            raise Exit(5, 'source_changed_during_run')
        row = rows[entry]
        if dst is not None:
            current, _ = data(TARGET, entry, code=3)
            if current != value:
                raise Exit(3, 'target_differs')
            row['outcome'] = 'already_identical'
            continue
        result = bao(['write', '-format=json', f'platform/data/{TARGET}/{entry}', '-'],
                     payload={'options': {'cas': 0}, 'data': value})
        if result.returncode != 0:
            raise Exit(4, 'write_failed')
        back, back_version = data(TARGET, entry, code=5)
        if back != value or back_version != 1:
            raise Exit(5, 'readback_mismatch')
        row['target_version'] = 1
        row['outcome'] = 'copied'

    for entry, src, _ in plan:
        if src is None:
            continue
        again = meta(SOURCE, entry)
        if again is None or again.get('current_version') != src['current_version']:
            raise Exit(5, 'source_changed_during_run')

    if args.keep_source:
        return 'copied_source_kept'

    # Phase 2: remove the undocumented infotech entries (all versions).
    for entry, src, _ in plan:
        if src is None:
            continue
        if bao(['delete', f'platform/metadata/{SOURCE}/{entry}']).returncode != 0:
            raise Exit(4, 'source_delete_failed')
        if meta(SOURCE, entry) is not None:
            raise Exit(5, 'source_still_present')
        rows[entry]['source_deleted'] = True
    return 'moved'


def main(argv=None):
    devnull = os.open(os.devnull, os.O_WRONLY)
    os.dup2(devnull, 1)
    os.dup2(devnull, 2)
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--confirm', default='')
    parser.add_argument('--receipt', type=Path)
    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('--keep-source', action='store_true')
    try:
        args = parser.parse_args(argv)
    except SystemExit:
        return 2
    receipt = {'schema': SCHEMA, 'mode': 'dry-run' if args.dry_run else 'move',
               'source': 'platform/' + SOURCE, 'target': 'platform/' + TARGET,
               'entries': {}, 'credential_values_emitted': False}
    fd = None
    if args.receipt is not None:
        try:
            fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        except OSError:
            return 2
    try:
        receipt['outcome'] = run(args, receipt)
        code = 0
    except Exit as exc:
        receipt['outcome'] = exc.reason
        code = exc.code
    except Exception:
        receipt['outcome'] = 'contained_internal_error'
        code = 1
    receipt['exit_code'] = code
    if fd is not None:
        os.write(fd, (json.dumps(receipt, indent=2, sort_keys=True) + '\n').encode())
        os.close(fd)
    return code


if __name__ == '__main__':
    sys.exit(main())