Le notebook matérialise la chronologie, montre quels enregistrements le moteur retourne aujourd’hui et lesquels sont hors intervalle, tente d’antidater une écriture et se fait refuser, applique la correction issue de l’image disque, provoque un VERSION_CONFLICT entre deux agents, reconstruit l’échelle des versions avec trace, trace le diagramme bitemporel, puis vérifie qu’un enregistrement mis en pierre tombale disparaît de recall tout en restant explicable.
temporal/notebook.ipynbEXTRAIT / 19 CELLULES
MARKDOWN [0]Reproduisez le scénario sur une copie de travail et conservez les identifiants de preuves, de session et de snapshot dans le résultat.
CODE [1] / PYTHONimport 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])
CODE [2] / PYTHONLIMITS = {
'max_items': 100,
'max_depth': 3,
'max_payload_bytes': 512 * 1024,
'max_cost': 2000,
'timeout_ms': 5000,
'supernode_threshold': 64,
}
visible = memory_op('recall', {
'objective': 'northwind',
'seed_ids': [ids['case--nw-supply']],
'limits': LIMITS,
})
returned = {item['record']['identity_key'] for item in visible['items']}
print(f"{'record':<34} {'valid window (days)':<22} retrieved")
print('-' * 70)
for key, (start, end) in sorted(valid_windows.items()):
window = f"{start:+d} .. {'open' if end is None else format(end, '+d')}"
print(f'{key:<34} {window:<22} {"yes" if key in returned else "no"}')
assert 'assessment--access-compromised' in returned
assert 'assessment--access-legitimate' not in returned # interval closed
assert 'contract--renewal' not in returned # not yet in force
assert 'obs--maintenance-notice' not in returned # expired
CODE [3] / PYTHONbackdated = dict(dataset['memories'][0])
backdated['identity_key'] = 'assessment--backdated-attempt'
backdated['valid_from'] = at(-30)
backdated['valid_until'] = None
backdated['recorded_at'] = at(-30) # the field a client wishes existed
refused = memory_op('remember', backdated, idempotency_key='temporal:backdate', expect_error=True)
print(refused.status, refused.code)
print(refused.message[:200])
assert refused.code == 'INVALID_REQUEST'
CODE [4] / PYTHONtarget = {'kind': 'memory', 'id': ids['assessment--access-compromised']}
before = memory_op('trace', {'target': target})
print('versions before:', [v['version'] for v in before['versions']])
corrected = memory_op('update', {
'target': target,
'expected_version': before['versions'][-1]['version'],
'patch': {
'confidence': 0.94,
'add_provenance': [{
'source_id': 'forensics',
'locator': 'image#build-03',
'observed_at': at(-4),
}],
'add_tags': ['forensically-confirmed'],
},
}, idempotency_key='temporal:correction')
record = corrected['record']
print('version :', record['version'])
print('confidence :', record['confidence'])
print('recorded_at:', record['recorded_at'])
print('provenance :', [p['source_id'] for p in record['provenance']])
print('valid_from :', record['valid_from'], '(unchanged: the fact was always true then)')
# The correction did not erase what justified the earlier, weaker belief.
assert len(record['provenance']) > 1
assert record['version'] > 1
CODE [5] / PYTHONladder = memory_op('trace', {'target': target})
for version in ladder['versions']:
print(f"v{version['version']} {version['recorded_at']} {version['lifecycle']}")
print()
print('actor :', ladder['actor_id'])
print('session :', ladder['session_id'])
print('policies :', ladder['policy_decisions'])
def state_known_at(versions, moment):
"""The version the system would have served at `moment`."""
seen = [v for v in versions if datetime.fromisoformat(v['recorded_at']) <= moment]
return seen[-1] if seen else None
first = datetime.fromisoformat(ladder['versions'][0]['recorded_at'])
print()
print('as the system stood at ingest :', state_known_at(ladder['versions'], first))
print('as the system stands now :', state_known_at(ladder['versions'], datetime.now(timezone.utc)))
assert state_known_at(ladder['versions'], first)['version'] == 1