Récupération bornée par working set

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

MAXIMAL = {
    'max_items': 10000,
    'max_depth': 6,
    'max_payload_bytes': 16 * 1024 * 1024,
    'max_cost': 1000000,
    'timeout_ms': 60000,
    'supernode_threshold': 100000,
}

wide = memory_op('recall', {
    'objective': 'atlas-07',
    'seed_ids': [ids['case--atlas-07']],
    'limits': MAXIMAL,
})

kinds = {}
for item in wide['items']:
    kinds[item['record']['kind']] = kinds.get(item['record']['kind'], 0) + 1

print('items       :', len(wide['items']))
print('relationships:', len(wide['relationships']))
print('completeness :', wide['completeness'])
print('usage        :', wide['usage'])
print('by kind      :', dict(sorted(kinds.items())))

BUDGETED = {
    'max_items': 60,
    'max_depth': 4,
    'max_payload_bytes': 512 * 1024,
    'max_cost': 5000,
    'timeout_ms': 5000,
    'supernode_threshold': 64,
}

focused = memory_op('recall', {
    'objective': 'atlas-07',
    'seed_ids': [ids['case--atlas-07']],
    'limits': BUDGETED,
})

print('items       :', len(focused['items']))
print('completeness :', focused['completeness'])
print('usage        :', focused['usage'])
print('recall_id    :', focused['recall_id'])
print('next page    :', focused['next_page_token'] is not None)

# Every selected record carries the reasons it entered the working set.
from collections import Counter

reasons = Counter()
for item in focused['items']:
    for reason in item['selection_reasons']:
        reasons[reason] += 1

for reason, count in reasons.most_common():
    print(f'{count:4d}  {reason}')

print()
for item in sorted(focused['items'], key=lambda i: -i['score'])[:8]:
    record = item['record']
    print(f"{item['score']:5.2f}  {record['identity_key']:<22} {','.join(item['selection_reasons'])}")

cheap = memory_op('recall', {
    'objective': 'atlas-07',
    'seed_ids': [ids['case--atlas-07']],
    'limits': dict(BUDGETED, max_cost=40),
})
print('cost-capped items:', len(cheap['items']), cheap['completeness']['outcomes'], cheap['usage'])
assert 'cost_budget_exhausted' in cheap['completeness']['outcomes']

# A zero budget is not 'unlimited'. It is invalid.
refused = memory_op('recall', {
    'objective': 'atlas-07',
    'seed_ids': [],
    'limits': dict(BUDGETED, max_depth=0),
}, expect_error=True)
print('rejected  :', refused.status, refused.code, '-', refused.message)
assert refused.code == 'INVALID_BUDGET'

rows = []
for depth in (1, 2, 3, 4, 5):
    for threshold in (32, 64, 128, 100000):
        limits = dict(BUDGETED, max_items=500, max_depth=depth, supernode_threshold=threshold)
        result = memory_op('recall', {
            'objective': 'atlas-07',
            'seed_ids': [ids['case--atlas-07']],
            'limits': limits,
        })
        campaign = sum(1 for i in result['items'] if 'atlas-07' in i['record']['tags'])
        rows.append({
            'depth': depth,
            'supernode_threshold': threshold,
            'items': len(result['items']),
            'campaign_items': campaign,
            'cost': result['usage']['cost'],
            'outcomes': ','.join(result['completeness']['outcomes']) or '-',
        })

header = f"{'depth':>5} {'threshold':>10} {'items':>6} {'campaign':>9} {'cost':>6}  outcomes"
print(header)
print('-' * len(header))
for row in rows:
    print(f"{row['depth']:>5} {row['supernode_threshold']:>10} {row['items']:>6}"
          f" {row['campaign_items']:>9} {row['cost']:>6}  {row['outcomes']}")

explanation = memory_op('trace', {'target': {'kind': 'recall', 'id': focused['recall_id']}})

print('actor   :', explanation['actor_id'])
print('agent   :', explanation['agent_id'])
print('session :', explanation['session_id'])
print('policies:', explanation['policy_decisions'])
print('details :', explanation['details'])
print()
for path in explanation['paths'][:5]:
    print('memories     :', path['memory_ids'][:4])
    print('relationships:', path['relationship_ids'][:4])
    print('evidence     :', path['evidence_source_ids'][:4])
    print()