Réponses porteuses de preuves

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': 3, 'max_payload_bytes': 1024 * 1024,
    'max_cost': 5000, 'timeout_ms': 10000, 'supernode_threshold': 64,
}


def build_envelope(recall, question):
    records = {i['record']['id']: i['record'] for i in recall['items']}
    buckets = {'supports': [], 'challenges': [], 'limits': []}
    for edge in recall['relationships']:
        if edge['kind'] in buckets and edge['source_id'] in records:
            buckets[edge['kind']].append(records[edge['source_id']])

    statement = next((r for r in records.values() if r['kind'] == 'analysis.statement'), None)
    provenance = sorted({p['source_id'] for r in records.values() for p in r['provenance']})

    return {
        'question': question,
        'statement': statement['content']['value']['text'] if statement else None,
        'confidence': statement['confidence'] if statement else None,
        'supporting_evidence': [
            {'id': r['identity_key'], 'reliability': r['content']['value']['properties'].get('source_reliability')}
            for r in buckets['supports']
        ],
        'counter_evidence': [
            {'id': r['identity_key'], 'reliability': r['content']['value']['properties'].get('source_reliability')}
            for r in buckets['challenges']
        ],
        'unresolved_unknowns': [
            {'id': r['identity_key'], 'impact': r['content']['value']['properties'].get('impact')}
            for r in buckets['limits']
        ],
        'source_provenance': provenance,
        'retrieval_completeness': {
            'complete': recall['completeness']['complete'],
            'truncated': recall['completeness']['truncated'],
            'outcomes': recall['completeness']['outcomes'],
            'usage': recall['usage'],
        },
        'recall_id': recall['recall_id'],
    }


full = memory_op('recall', {
    'objective': 'kestrel',
    'seed_ids': [ids['question--kestrel-attribution']],
    'limits': LIMITS,
})
envelope = build_envelope(full, 'Is the intrusion attributable to Kestrel?')
print(json.dumps(envelope, indent=2))

# An answer that cannot name what contradicts it is not an answer, it is a claim.
assert envelope['counter_evidence'], 'the envelope must carry counter-evidence'
assert envelope['unresolved_unknowns'], 'the envelope must carry its gaps'

print('supporting :', len(envelope['supporting_evidence']))
print('challenging:', len(envelope['counter_evidence']))
print('unknowns   :', len(envelope['unresolved_unknowns']))
print()

# The strongest-sounding support is the weakest source in the set.
for item in sorted(envelope['supporting_evidence'], key=lambda e: e['reliability']):
    print(f"  supports   {item['id']:<32} {item['reliability']}")
for item in envelope['counter_evidence']:
    print(f"  challenges {item['id']:<32} {item['reliability']}")

# Seed the statement explicitly so it is guaranteed to be in the page: the
# point is to lose the *argument*, not the conclusion.
starved = memory_op('recall', {
    'objective': 'kestrel',
    'seed_ids': [ids['statement--attribution-plausible']],
    'limits': dict(LIMITS, max_items=3, max_cost=4),
})
partial = build_envelope(starved, 'Is the intrusion attributable to Kestrel?')

print(f"{'':<22}{'full':>10}{'starved':>12}")
print('-' * 44)
print(f"{'confidence':<22}{str(envelope['confidence']):>10}{str(partial['confidence']):>12}")
print(f"{'supporting':<22}{len(envelope['supporting_evidence']):>10}{len(partial['supporting_evidence']):>12}")
print(f"{'counter-evidence':<22}{len(envelope['counter_evidence']):>10}{len(partial['counter_evidence']):>12}")
print(f"{'unknowns':<22}{len(envelope['unresolved_unknowns']):>10}{len(partial['unresolved_unknowns']):>12}")
print(f"{'complete':<22}{str(envelope['retrieval_completeness']['complete']):>10}"
      f"{str(partial['retrieval_completeness']['complete']):>12}")
print(f"{'truncated':<22}{str(envelope['retrieval_completeness']['truncated']):>10}"
      f"{str(partial['retrieval_completeness']['truncated']):>12}")
print(f"{'items examined':<22}{envelope['retrieval_completeness']['usage']['items']:>10}"
      f"{partial['retrieval_completeness']['usage']['items']:>12}")

# Same conclusion, same confidence, a fraction of the argument behind it.
assert partial['confidence'] == envelope['confidence']
assert partial['retrieval_completeness']['truncated'] is True
assert len(partial['counter_evidence']) < len(envelope['counter_evidence'])