AIOS Governed Continuation
591884bf5f
autonomy-smoke-tests phase1-p0 PASS evidence_sha256=d3ed00952a599a4d17d6cf5e8f9714c233ada76347b3567acab8455fb07c7717
autonomy-unit-tests phase1-p0 PASS evidence_sha256=77c39918bb5883e62fbbae39f1812a1b5b6e19c98ba422d988ab509f84575b35
campaign-approval-gate phase1-p0 PASS evidence_sha256=8f562805faf8d739233301596d3a95d6a128eff7324fd30d663d94a196cf5c13
git-diff-check phase1-p0 PASS evidence_sha256=dc18ae7a546ada66b0bfdd9045464f8413312234baf8d3004eac78387c614134
py-compile phase1-p0 PASS evidence_sha256=ed151cbcc033e00f78a86e86840883d5996b2c7a33ed4520421cc3a6f4423dd3
reconciliation-contract phase1-p0 PASS evidence_sha256=3a4243f272ab2458dc631eb751f267f93c48713d96a59b311ae885e4d0f70728
release-state-machine phase1-p0 PASS evidence_sha256=d24c81a9969c35fc4c599195e3078c10fb06a0b6addfed9b44f4972b228464b9
secret-scan phase1-p0 PASS evidence_sha256=649005c3fef5220b20c9f616658ad283a28d506fe231d8bb6f795de183ad51c5
signer-validator-contract phase1-p0 PASS evidence_sha256=b5461be6646b4347c1ad85cf13889eb03de515ce5e89c041a8f24626fae2ec17
33 lines
1.0 KiB
Python
33 lines
1.0 KiB
Python
"""Collision-resistant storage keys for durable governed Task state."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
|
|
|
|
class JobIdentityError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def legacy_task_storage_key(task_id: str) -> str:
|
|
"""Return the historical lossy key solely to detect migration conflicts."""
|
|
if not isinstance(task_id, str) or not task_id:
|
|
raise JobIdentityError("TASK_ID_INVALID")
|
|
return re.sub(r"[^A-Za-z0-9_.-]+", "-", task_id).strip("-")[:120] or "task"
|
|
|
|
|
|
def task_storage_key(task_id: str) -> str:
|
|
"""Return a canonical collision-resistant key for every Task state sink."""
|
|
if (
|
|
not isinstance(task_id, str)
|
|
or not task_id
|
|
or task_id != task_id.strip()
|
|
or len(task_id) > 512
|
|
):
|
|
raise JobIdentityError("TASK_ID_INVALID")
|
|
readable = re.sub(r"[^A-Za-z0-9_.-]+", "-", task_id).strip("-")[:48] or "task"
|
|
digest = hashlib.sha256(
|
|
("aios-task-storage-v1\0" + task_id).encode("utf-8")
|
|
).hexdigest()
|
|
return f"{readable}-{digest}"
|