secrets-engine/tools/prepare_metered_activation.py

"""Offline review packet for a metered owner; never requests approval or credentials.

Run with the owning secrets-engine environment. The provider facts input is a
reviewed snapshot, not authority, a live price feed, or a token estimate.
"""
from __future__ import annotations

import argparse
from dataclasses import asdict
from decimal import Decimal
import hashlib
import json
from pathlib import Path
from types import SimpleNamespace

from secrets_engine.approval_consume import _expected_request
from secrets_engine.catalog import get_entry
from secrets_engine.exec_owner import owner_binding, owner_digest, resolve_companions
from secrets_engine.plan import build_plan


def review_policy(owner: dict, spend: dict, provider: dict) -> dict:
    policy = owner['messages_policy']
    failures = []
    for key in ('context_tokens', 'max_output_tokens', 'input_microusd_per_token',
                'output_microusd_per_token'):
        if type(policy.get(key)) is not int or policy[key] <= 0:
            raise ValueError('invalid_policy_bound')
    for key in ('context_tokens', 'max_output_tokens', 'input_microusd_per_token',
                'output_microusd_per_token'):
        if type(provider.get(key)) is not int or provider[key] <= 0:
            raise ValueError('invalid_provider_bound')
    if policy['model'] != provider['model']:
        failures.append('model_mismatch')
    # No tokenizer or count endpoint is used by MessagesPolicy. A smaller
    # configured context does not constrain what the upstream model accepts.
    if policy['context_tokens'] < provider['context_tokens']:
        failures.append('input_reservation_below_provider_context')
    if policy['max_output_tokens'] > provider['max_output_tokens']:
        failures.append('output_exceeds_provider_limit')
    observed_output = provider.get('observed_cli_max_tokens')
    if observed_output is not None:
        if type(observed_output) is not int or observed_output <= 0:
            raise ValueError('invalid_observed_output_bound')
        if policy['max_output_tokens'] < observed_output:
            failures.append('output_limit_below_observed_cli_request')
    required_betas = provider.get('required_request_betas', [])
    if set(required_betas) - set(policy.get('allowed_betas', [])):
        failures.append('observed_cli_beta_not_admitted')
    for key in ('input_microusd_per_token', 'output_microusd_per_token'):
        if policy[key] < provider[key]:
            failures.append('understated_' + key)
    liability = (policy['context_tokens'] * policy['input_microusd_per_token']
                 + policy['max_output_tokens'] * policy['output_microusd_per_token'])
    capacity = Decimal(spend['max_liability_usd']) * 1_000_000
    fx = Decimal(spend['eur_per_usd'])
    cap = Decimal(spend['per_run_eur'])
    if not all(x.is_finite() and x > 0 for x in (capacity, fx, cap)):
        raise ValueError('invalid_spend_bound')
    if capacity * fx > cap * 1_000_000:
        failures.append('parent_exceeds_eur_cap')
    if liability > capacity:
        failures.append('no_full_output_request_fits')
    return {
        'passed': not failures, 'failures': failures,
        'maximum_request_microusd': liability,
        'maximum_request_eur_at_configured_fx': str(Decimal(liability) * fx / 1_000_000),
        'full_output_requests_that_fit': int(capacity // liability),
        'provider_facts_ref': provider['source'],
        'authority_granted': False,
        'remaining_acceptance': ['FX upper bound and validity', 'live CLI/beta compatibility',
                                 'native delivery and runtime admission'],
    }


def build_packet(catalog_dir: Path, primary_id: str, owner: dict, spend: dict,
                 provider: dict) -> dict:
    primary = get_entry(catalog_dir, primary_id)
    binding = owner_binding(primary)
    if binding is None or binding.get('status') != 'configured':
        raise ValueError('configured_owner_required')
    # This is a source/offline review: host file validation is a separate,
    # explicit backend-free step. Do not pretend local paths validate a host.
    lanes = [(primary, primary.fields[0])]
    lanes.extend((lane, field) for lane, field, _ in resolve_companions(
        primary, lambda cid: get_entry(catalog_dir, cid)))
    cfg = SimpleNamespace(authorization_subject_id='secrets-engine',
                          authorization_subject_type='service')
    actions = []
    for lane, field in lanes:
        for action in ('apply', 'verify', 'exec'):
            request = _expected_request(
                cfg, lane, action,
                fields=() if action == 'apply' else tuple(lane.fields) if action == 'verify' else (field,),
                policy_targets=(lane.policy_name,), auth_targets=(lane.role_name,))
            actions.append({'catalog': lane.id, 'action': action, 'request': request})
    return {
        'schema': 'metered-activation-review/v1', 'status': 'review-only',
        'dispatch_enabled': False, 'owner_digest': owner_digest(primary),
        'policy_review': review_policy(owner, spend, provider),
        'plans': [asdict(build_plan(lane, lane.stage)) for lane, _ in lanes],
        'actions': actions,
        'native_evaluator_digests': 'not obtained; local hashes are not evaluator digests',
        'approval_ids': [], 'host_pins_verified': False,
    }


def load_pinned_snapshot(path: Path, expected: str) -> dict:
    raw = path.read_bytes()
    if hashlib.sha256(raw).hexdigest() != expected:
        raise ValueError('snapshot_file_pin_mismatch')
    return json.loads(raw)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--catalog-dir', type=Path, required=True)
    parser.add_argument('--primary', default='glas-claude-agent-dev-anthropic')
    parser.add_argument('--owner-snapshot', type=Path, required=True)
    parser.add_argument('--spend-snapshot', type=Path, required=True)
    parser.add_argument('--provider-facts', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    binding = owner_binding(get_entry(args.catalog_dir, args.primary))
    if binding is None or binding.get('status') != 'configured':
        raise ValueError('configured_owner_required')
    command = binding['command']
    owner_path = command[command.index('--owner-config') + 1]
    spend_path = binding['environment']['AGENT_HARNESS_SPEND_POLICY']
    owner = load_pinned_snapshot(args.owner_snapshot, binding['files'][owner_path]['sha256'])
    spend = load_pinned_snapshot(args.spend_snapshot, binding['files'][spend_path]['sha256'])
    packet = build_packet(args.catalog_dir, args.primary, owner, spend,
                          json.loads(args.provider_facts.read_text()))
    # Refuse overwrite: each revision remains independently reviewable.
    with args.output.open('x') as out:
        json.dump(packet, out, indent=2)
        out.write('\n')
    return 0 if packet['policy_review']['passed'] else 2


if __name__ == '__main__':
    raise SystemExit(main())