Mondes hypothétiques parallèles

import json, os, pathlib, time
import requests
import urllib3

# The quick-start server uses a self-signed certificate; disable verification
# for the local playbook only, never against a real deployment.
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

BASE_URL = os.environ.get('CORROBORE_URL', 'https://127.0.0.1:8080')
TOKEN = os.environ.get('CORROBORE_HTTP_AUTH_TOKEN', 'change-me')

http = requests.Session()
http.verify = False
http.headers.update({'Authorization': f'Bearer {TOKEN}', 'Content-Type': 'application/json'})


class MemoryError_(RuntimeError):
    """Carries the stable v1 error taxonomy instead of a bare HTTP status."""

    def __init__(self, code, message, status):
        super().__init__(f'{code}: {message}')
        self.code, self.message, self.status = code, message, status


def memory_op(operation, payload, idempotency_key=None, expect_error=False):
    """POST /v1/memory/operations and unwrap the typed result."""
    body = {'contract_version': 'v1', 'operation': operation, 'input': payload}
    if idempotency_key is not None:
        body['idempotency_key'] = idempotency_key
    for attempt in range(12):
        response = http.post(f'{BASE_URL}/v1/memory/operations', data=json.dumps(body))
        # Protected routes share a global token bucket (50 rps sustained, 200 burst by
        # default) that a bulk load will hit. Two details matter here: the 429 body is
        # plain text, not the JSON error envelope, and `Retry-After` can be `0` — so
        # honour it as a floor, never as the whole wait.
        if response.status_code == 429:
            hinted = float(response.headers.get('Retry-After', 0) or 0)
            time.sleep(max(hinted, 0.2 * (attempt + 1)))
            continue
        break
    if response.status_code != 200:
        try:
            error = response.json().get('error', {})
        except ValueError:  # 429 and other transport rejections are not JSON
            error = {}
        failure = MemoryError_(error.get('code', 'UNKNOWN'), error.get('message', response.text), response.status_code)
        if expect_error:
            return failure
        raise failure
    if expect_error:
        raise AssertionError(f'{operation} unexpectedly succeeded')
    return response.json()['result']['result']


ready = http.get(f'{BASE_URL}/health/ready').json()
version = http.get(f'{BASE_URL}/version').json()
print('ready  :', json.dumps(ready)[:160])
print('version:', json.dumps(version)[:160])

assumed = {}
for edge in edges:
    if edge['kind'] == 'assumes':
        assumed.setdefault(records[edge['source_id']]['identity_key'], set()).add(
            records[edge['target_id']]['identity_key'])

worlds = sorted(assumed)
print('worlds:', worlds)
for name in worlds:
    print(f'  {name} assumes {len(assumed[name])} base facts')

shared = set.intersection(*assumed.values())
print()
print('base facts shared by both branches:', len(shared))
for fact in sorted(shared):
    print(f'  {fact}')

assert len(shared) == dataset['facts']['shared_base_facts']
assert all(assumed[name] == shared for name in worlds), 'neither branch has private facts'

predicts = {}
for edge in edges:
    if edge['kind'] == 'predicts':
        predicts.setdefault(records[edge['source_id']]['identity_key'], set()).add(
            records[edge['target_id']]['identity_key'])

for name in worlds:
    print(name)
    for prediction in sorted(predicts[name]):
        print(f'    {prediction}')

both = predicts[worlds[0]] & predicts[worlds[1]]
print()
print('predicted by BOTH branches:', sorted(both))
print('    testing these can never separate the two explanations')

outcomes = {}
for edge in edges:
    if edge['kind'] in ('confirms', 'refutes'):
        observation = records[edge['source_id']]['identity_key']
        outcomes.setdefault(observation, []).append(
            (edge['kind'], records[edge['target_id']]['identity_key']))


def separates(observation):
    """An observation discriminates when its effects fall on different branches."""
    touched = set()
    for _, prediction in outcomes.get(observation, []):
        touched |= {w for w in worlds if prediction in predicts[w]}
    exclusive = {p for _, p in outcomes.get(observation, []) if p not in both}
    return len(touched) > 1 and bool(exclusive)


print(f"{'observation':<34}{'discriminates':>14}   effects")
print('-' * 92)
for observation in sorted(outcomes):
    effects = ', '.join(f'{kind} {prediction}' for kind, prediction in sorted(outcomes[observation]))
    print(f'{observation:<34}{str(separates(observation)):>14}   {effects}')

trap = 'observation--account-shared'
assert separates(trap) is False, 'both worlds predicted the shared account'
assert separates('observation--asn-mismatch') is True

def score_worlds(exclude_non_discriminating=True):
    tally = {name: {'confirmed': 0, 'refuted': 0} for name in worlds}
    for observation, effects in outcomes.items():
        if exclude_non_discriminating and not separates(observation):
            continue
        for kind, prediction in effects:
            for name in worlds:
                if prediction in predicts[name]:
                    tally[name]['confirmed' if kind == 'confirms' else 'refuted'] += 1
    return tally


for label, strict in (('all observations', False), ('discriminating only', True)):
    tally = score_worlds(strict)
    print(label)
    for name in worlds:
        print(f"    {name:<20} confirmed {tally[name]['confirmed']}  refuted {tally[name]['refuted']}")

strict = score_worlds(True)
leader = max(worlds, key=lambda w: strict[w]['confirmed'] - strict[w]['refuted'])
print()
print('branch ahead on discriminating evidence:', leader)
assert leader == 'world--supplier'

world_ids = [ids[name] for name in worlds]
CONSOLIDATION = {
    'memory_ids': world_ids,
    'canonical_id': ids[leader],
    'reason': 'supplier branch leads on discriminating evidence; insider branch retained',
    'preserve_disagreements': True,
}

proposal = memory_op('consolidate', dict(CONSOLIDATION, mode={'mode': 'propose'}))
print('proposal :', proposal['proposal_id'])
print('applied  :', proposal['applied'], '(a proposal changes nothing)')

erased = memory_op('consolidate', dict(CONSOLIDATION,
                                       mode={'mode': 'apply_approved',
                                             'proposal_id': proposal['proposal_id'],
                                             'approval_policy': 'policy--lead-analyst'},
                                       preserve_disagreements=False),
                   idempotency_key='pw:erase', expect_error=True)
print()
print('closing the case by erasing the loser:', erased.status, erased.code)
print(' ', erased.message)
assert erased.code == 'POLICY_APPROVAL_REQUIRED'