Multi-résolution et hyperrelations

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])

BUDGET = {
    'max_items': 500, 'max_depth': 3, 'max_payload_bytes': 4 * 1024 * 1024,
    'max_cost': 20000, 'timeout_ms': 20000, 'supernode_threshold': 512,
}

EPISODE_KEYS = ['episode--seed', 'episode--amplify-1', 'episode--localise',
                'episode--authority', 'episode--amplify-2', 'episode--decay']


def ask(objective, seeds, limits=BUDGET):
    if isinstance(seeds, str):
        seeds = [seeds]
    return memory_op('recall', {
        'objective': objective,
        'seed_ids': [ids[s] for s in seeds],
        'limits': limits,
    })


raw = ask('tidewater', EPISODE_KEYS, dict(BUDGET, max_depth=1))
coarse = ask('tidewater', 'campaign--tidewater', dict(BUDGET, max_depth=1))

for label, result in (('raw posts', raw), ('campaign view', coarse)):
    kinds = {}
    for item in result['items']:
        kinds[item['record']['kind']] = kinds.get(item['record']['kind'], 0) + 1
    print(f"{label:<16} items {len(result['items']):>4}  cost {result['usage']['cost']:>5}"
          f"  bytes {result['usage']['payload_bytes']:>7}  {dict(sorted(kinds.items()))}")

print()
print(f"payload ratio: {raw['usage']['payload_bytes'] / coarse['usage']['payload_bytes']:.1f}x")

assert len(coarse['items']) < len(raw['items'])
assert coarse['usage']['payload_bytes'] < raw['usage']['payload_bytes']

def open_episode(key):
    """Return the posts a single episode was derived from."""
    opened = ask('derivation', key, dict(BUDGET, max_depth=1))
    records = {i['record']['id']: i['record'] for i in opened['items']}
    origin = ids[key]
    posts = [records[e['target_id']]['identity_key']
             for e in opened['relationships']
             if e['kind'] == 'derived_from' and e['source_id'] == origin
             and e['target_id'] in records]
    return posts, opened['usage']


posts, usage = open_episode('episode--amplify-1')
print('episode        : episode--amplify-1')
print('posts behind it:', len(posts))
print('sample         :', sorted(posts)[:6])
print('cost to open   :', usage['cost'])

# Every episode opens onto its own slice, and the slices partition the corpus.
print()
total = 0
for key in EPISODE_KEYS:
    backing, _ = open_episode(key)
    total += len(backing)
    print(f'{key:<22} opens onto {len(backing):>3} source posts')
    assert backing, f'{key} has no path back to its evidence'

print(f'\ntotal reachable posts: {total} of {dataset["facts"]["raw_posts"]}')
assert total == dataset['facts']['raw_posts']

burst = next(r['record'] for r in ask('coordination', 'event--four-minute-burst',
                                      dict(BUDGET, max_depth=1))['items']
             if r['record']['identity_key'] == 'event--four-minute-burst')
props = burst['content']['value']['properties']
participants = props['participants']

print('window     :', props['window_start'], '->', props['window_end'])
print('participants:', props['participant_count'])
for account, role in sorted(participants.items()):
    print(f'    {account}  {role}')

n = props['participant_count']
pairwise = n * (n - 1) // 2
print()
print(f'as one n-ary event : 1 record, {n} roles')
print(f'as pairwise edges  : {pairwise} edges, and no role survives')
assert pairwise == 28