Phéromones et anti-phéromones

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


def traverse(objective, seed, limits=LIMITS):
    recall = memory_op('recall', {
        'objective': objective,
        'seed_ids': [ids[seed]],
        'limits': limits,
    })
    records = {i['record']['id']: i['record'] for i in recall['items']}
    evidence = sum(1 for r in records.values() if r['kind'] == 'retrieval.evidence')
    return recall, records, evidence


recall, records, evidence = traverse('beneficiary financing', 'case--fimi-01')
print('items     :', len(recall['items']))
print('cost      :', recall['usage']['cost'])
print('outcomes  :', recall['completeness']['outcomes'])
print('evidence  :', evidence)
print('edge kinds:', sorted({e['kind'] for e in recall['relationships']}))

from collections import defaultdict

DECAY = 0.85


class NavigationField:
    """Positive and negative traces per edge kind, scoped to one task family."""

    def __init__(self, scope):
        self.scope = scope
        self.positive = defaultdict(float)
        self.negative = defaultdict(float)

    def decay(self):
        for table in (self.positive, self.negative):
            for kind in list(table):
                table[kind] *= DECAY

    def observe(self, recall, records):
        self.decay()
        cost = max(recall['usage']['cost'], 1)
        evidence_ids = {i for i, r in records.items() if r['kind'] == 'retrieval.evidence'}
        barren_ids = {i for i, r in records.items() if r['kind'] == 'retrieval.restatement'}
        for edge in recall['relationships']:
            kind = edge['kind']
            if edge['target_id'] in evidence_ids:
                self.positive[kind] += 1.0 / cost * 100
            if edge['target_id'] in barren_ids:
                self.negative[kind] += 1.0 / cost * 100
        # Hitting a bound is a property of the path, so it repels too.
        for outcome in recall['completeness']['outcomes']:
            for edge in recall['relationships']:
                self.negative[edge['kind']] += 0.05

    def score(self, kind):
        return round(self.positive[kind] - self.negative[kind], 3)

    def ranked(self):
        kinds = set(self.positive) | set(self.negative)
        return sorted(((k, self.score(k)) for k in kinds), key=lambda kv: -kv[1])


fimi = NavigationField('task--fimi-financing')
for _ in range(6):
    recall, records, _ = traverse('beneficiary financing', 'case--fimi-01')
    fimi.observe(recall, records)

print(f"{'edge kind':<26}{'score':>8}{'positive':>10}{'negative':>10}")
print('-' * 54)
for kind, score in fimi.ranked():
    print(f'{kind:<26}{score:>8}{fimi.positive[kind]:>10.2f}{fimi.negative[kind]:>10.2f}')

attractive = [k for k, s in fimi.ranked() if s > 0]
repelled = [k for k, s in fimi.ranked() if s < 0]
print()
print('attractive:', attractive)
print('repelled  :', repelled)

assert 'documented_by' in attractive, 'the path to filings should gain utility'
assert 'restated_by' in repelled, 'the path to restatements should be penalised'

cti = NavigationField('task--cti-infrastructure')
for _ in range(6):
    recall, records, _ = traverse('command host intrusion', 'case--cti-01')
    cti.observe(recall, records)

print(f"{'task':<28}{'top edge kind':<24}{'score':>8}")
print('-' * 62)
for field in (fimi, cti):
    kind, score = field.ranked()[0]
    print(f'{field.scope:<28}{kind:<24}{score:>8}')

print()
print('fimi knows about  :', sorted(set(fimi.positive) | set(fimi.negative)))
print('cti knows about   :', sorted(set(cti.positive) | set(cti.negative)))

# Neither field has learned anything about the other family's edges.
assert 'yielded' not in fimi.positive
assert 'documented_by' not in cti.positive
assert fimi.ranked()[0][0] != cti.ranked()[0][0]

best_kind, best_score = fimi.ranked()[0]
print(f'most reinforced edge kind: {best_kind} (score {best_score})')

starved, _, _ = traverse('beneficiary financing', 'case--fimi-01',
                         dict(LIMITS, max_cost=8, max_items=5))
print()
print('items    :', len(starved['items']))
print('cost     :', starved['usage']['cost'], 'of a budget of 8')
print('outcomes :', starved['completeness']['outcomes'])
print('complete :', starved['completeness']['complete'])

# No amount of accumulated utility buys extra traversal.
assert starved['usage']['cost'] <= 8
assert starved['completeness']['complete'] is False
assert len(starved['items']) < len(recall['items'])