Coopération épistémique multi-agent

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

LIMITS = {
    'max_items': 200,
    'max_depth': 4,
    'max_payload_bytes': 1024 * 1024,
    'max_cost': 5000,
    'timeout_ms': 10000,
    'supernode_threshold': 64,
}

state = memory_op('recall', {
    'objective': 'attribution',
    'seed_ids': [ids['case--attribution']],
    'limits': LIMITS,
})

records = {item['record']['id']: item['record'] for item in state['items']}
positions = {}
for edge in state['relationships']:
    if edge['kind'] in ('supports', 'challenges', 'abstains'):
        claim = records[edge['target_id']]['identity_key']
        positions.setdefault(claim, []).append((edge['properties']['agent'], edge['kind'], edge['confidence']))

for claim, held in sorted(positions.items()):
    print(claim)
    for agent, kind, confidence in sorted(held):
        print(f'    {agent:<20} {kind:<12} {confidence:.2f}')

assert len(positions) == 3, 'all three claims survive retrieval'

claim_ids = [ids[key] for key in ('claim--atlas-owns', 'claim--reseller-owns', 'claim--unresolved')]

# The canonical choice is part of what gets approved, so it belongs in the
# proposal. The analyst's position is that the honest answer is 'unresolved'.
CONSOLIDATION = {
    'memory_ids': claim_ids,
    'canonical_id': ids['claim--unresolved'],
    'reason': 'no primary log, ownership stays open',
    'preserve_disagreements': True,
}

proposal = memory_op('consolidate', dict(CONSOLIDATION, mode={'mode': 'propose'}))
print(json.dumps(proposal, indent=2))

assert proposal['applied'] is False
assert proposal['receipt'] is None

# Nothing moved: every claim is still active at version 1.
for key in ('claim--atlas-owns', 'claim--reseller-owns', 'claim--unresolved'):
    ladder = memory_op('trace', {'target': {'kind': 'memory', 'id': ids[key]}})
    assert [(v['version'], v['lifecycle']) for v in ladder['versions']] == [(1, 'active')]
print('\nno memory was modified by the proposal')

APPROVED = {
    'mode': 'apply_approved',
    'proposal_id': proposal['proposal_id'],
    'approval_policy': 'policy--analyst-review',
}

refused = memory_op('consolidate', dict(CONSOLIDATION,
                                        mode=APPROVED,
                                        preserve_disagreements=False,
                                        reason='collapse to the most confident claim'),
                    idempotency_key='epistemic:destructive', expect_error=True)

print(refused.status, refused.code)
print(refused.message)
assert refused.code == 'POLICY_APPROVAL_REQUIRED'

swapped = memory_op('consolidate', dict(CONSOLIDATION,
                                        mode=APPROVED,
                                        canonical_id=ids['claim--atlas-owns']),
                    idempotency_key='epistemic:swapped', expect_error=True)

print(swapped.status, swapped.code, '-', swapped.message)
assert swapped.code == 'POLICY_APPROVAL_REQUIRED'

# Quietly dropping a dissenting claim from the candidate set is refused too.
narrowed = memory_op('consolidate', dict(CONSOLIDATION,
                                         mode=APPROVED,
                                         memory_ids=claim_ids[1:]),
                     idempotency_key='epistemic:narrowed', expect_error=True)
print(narrowed.status, narrowed.code, '-', narrowed.message)
assert narrowed.code == 'POLICY_APPROVAL_REQUIRED'

# Input validation runs first: a canonical outside the candidate set never
# reaches the approval check at all.
orphaned = memory_op('consolidate', dict(CONSOLIDATION,
                                         mode=APPROVED,
                                         memory_ids=claim_ids[:2]),
                     idempotency_key='epistemic:orphaned', expect_error=True)
print(orphaned.status, orphaned.code, '-', orphaned.message)
assert orphaned.code == 'INVALID_REQUEST'

applied = memory_op('consolidate', dict(CONSOLIDATION, mode=APPROVED),
                    idempotency_key='epistemic:apply')

print('applied      :', applied['applied'])
print('canonical    :', by_id[applied['canonical_id']])
print('retained     :', [by_id[i] for i in applied['originals_retained']])
print('disagreements:', applied['disagreements_retained'])
print('receipt      :', applied['receipt'])

assert applied['disagreements_retained'] is True
assert len(applied['originals_retained']) == 3

after = memory_op('recall', {
    'objective': 'attribution',
    'seed_ids': [ids['case--attribution']],
    'limits': LIMITS,
})
returned = {item['record']['identity_key'] for item in after['items']}

print('claims still retrieved:', sorted(k for k in returned if k.startswith('claim--')))
print()
for key in ('claim--atlas-owns', 'claim--reseller-owns', 'claim--unresolved'):
    ladder = memory_op('trace', {'target': {'kind': 'memory', 'id': ids[key]}})
    print(f"{key:<24} {[(v['version'], v['lifecycle']) for v in ladder['versions']]}")

# The superseded readings are gone from retrieval and intact in the audit.
assert 'claim--unresolved' in returned
assert 'claim--atlas-owns' not in returned