Compare commits

..
Author SHA1 Message Date
AIOS Governed Continuation 591884bf5f fix(governed): provide collision-safe task storage identity
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
2026-09-07 21:32:46 +08:00
AIOS Governed Continuation 5cf599e1d1 feat: integrate formal lineage supersession P0 2026-09-07 21:28:09 +08:00
owner b3cee6ae48 Merge pull request 'feat(discovery): governed immutable repository discovery P0' (#73) from agent/repository-discovery-p0-20260902 into main
Reviewed-on: #73
2026-09-07 20:43:02 +08:00
AIOS Governed Continuation c8c25fa672 fix(discovery): make P0 contract hermetic
autonomy-smoke-tests phase1-p0 PASS evidence_sha256=13aff2dfe664be536470ca26a6ac3f20ba15fdca37cd4f8e799c0c8c36974be9
autonomy-unit-tests phase1-p0 PASS evidence_sha256=118708029df05d861585179a31e4f4eaf0d7953563f5347b76732f2055785ba0
campaign-approval-gate phase1-p0 PASS evidence_sha256=4e39d586b32ad2d17da8fc38f5d359ace8a239af8687b4f894a338fe9f0bf83c
git-diff-check phase1-p0 PASS evidence_sha256=2638b909fa281012b6a43aaf34e1d611b0dba1fc78bf47f54941739973ee6281
py-compile phase1-p0 PASS evidence_sha256=440cf54c4225f809a575eff9b24238acd361497fd18bffbb359087e88310dcd6
reconciliation-contract phase1-p0 PASS evidence_sha256=131fb1042ff1ceb565dbf1ec7aed1cd3879d728f912745c0589a02b96381baaf
release-state-machine phase1-p0 PASS evidence_sha256=a74b5192b248b6e342afeea01c85ecbb8a02d71891bcf334df2e89d96cc1577d
secret-scan phase1-p0 PASS evidence_sha256=d4789e5c45ba681c651c613427d70d911a4cdd70fbead0772c4e1e376bef41cd
signer-validator-contract phase1-p0 PASS evidence_sha256=a29c7664de7aa9e30e966f8c4a14c0eaf8191fbb92ba726a72ba3c7ec2659eec
2026-09-07 20:33:11 +08:00
AIOS Governed Continuation 1735371e06 feat(discovery): integrate P0 discovery with task admission bootstrap
autonomy-smoke-tests phase1-p0 PASS evidence_sha256=a96e50527a39e969b53df42ce799526faa8e3bcd618b6f5e2ec2e2e13c574f02
autonomy-unit-tests phase1-p0 FAIL evidence_sha256=d61da9d108d6ace7d845f78b1010773b4b1ce315a538209d7e60eb870c8eb953
campaign-approval-gate phase1-p0 PASS evidence_sha256=29d249ebfb389c3137e411c1d87fb88c7f755cdba4fea5118be187259c707295
git-diff-check phase1-p0 FAIL evidence_sha256=a0adbb54c3b220b9ffc82656470f1d67904823e461fb8ffedd713846814acc78
py-compile phase1-p0 PASS evidence_sha256=f504e34aa0a28ad009ba3b3fd3a75448297be43c452f4dc5ba445d3a4d0f7f07
reconciliation-contract phase1-p0 PASS evidence_sha256=8965ee765da46e32e79d046eeb57db023a2e9bd5105f057a27546b42eed470e3
release-state-machine phase1-p0 PASS evidence_sha256=e5a4c127d4b5b6eb27918fd9c54c06ee23d36d613bb816be475006c81b8ff479
secret-scan phase1-p0 PASS evidence_sha256=fc82389ca65008f045aa9357e5f276eeb0ce040ac5511197d65f1027975bc238
signer-validator-contract phase1-p0 PASS evidence_sha256=79a1d0246c73f0485816f77345fbc769c23f659cd177fefba90bb27e842214f3
2026-09-07 20:14:22 +08:00
AIOS Governed Continuation 1aee083564 feat: formal lineage supersession P0 2026-09-07 08:27:33 +08:00
AIOS Governed Continuation 1efa7724c1 feat(discovery): governed immutable repository discovery P0 2026-09-04 00:16:46 +08:00
34 changed files with 9226 additions and 39 deletions
@@ -0,0 +1,122 @@
"""Staging HTTP overlay for the exact Task lineage supersession capability."""
from __future__ import annotations
from http.server import ThreadingHTTPServer
from typing import Any, Callable, Mapping
from urllib.parse import urlparse
from aios_task_lineage_supersession import (
DECISION_SCHEMA,
TaskLineageSupersessionController,
TaskLineageSupersessionError,
)
SUPERSESSION_ROUTE = "/v1/control/task-lineage-supersession"
def _status(error: TaskLineageSupersessionError) -> int:
return error.http_status
class LineageSupersessionEndpoint:
"""Framework-neutral endpoint; authentication must precede all parsing/mutation."""
def __init__(
self,
db_path,
verifier,
authenticate_control: Callable[[Mapping[str, str]], Any],
*,
clock=None,
identity_digest=None,
) -> None:
self.authenticate_control = authenticate_control
self.controller = TaskLineageSupersessionController(
db_path, verifier, clock=clock, identity_digest=identity_digest
)
def post(self, headers: Mapping[str, str], body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
self.authenticate_control(headers)
try:
result = self.controller.supersede(body)
except TaskLineageSupersessionError as exc:
return _status(exc), {"error": exc.code}
return (200 if result.get("idempotent") else 201), result
def build_runtime():
"""Bind to the already-active recovery API and its formal Owner verifier lazily."""
try:
from aios_agent_task_api import APIConfig
from aios_agent_task_api_p0_failed_lease_recovery import (
P0FailedLeaseRecoveryHandler,
P0FailedLeaseRecoveryTaskApplication,
)
from aios_agent_task_protocol import TaskProtocolError
from aios_owner_approval_verifier import (
GrantRootOwnerApprovalVerifier,
OwnerApprovalVerifierError,
)
except ImportError as exc:
raise RuntimeError("LINEAGE_SUPERSESSION_RUNTIME_OVERLAY_UNAVAILABLE") from exc
# The active entrypoint layers failed- and active-lease recovery on top of
# expired recovery. Inherit that current layer so adding lineage does not
# silently remove either already-deployed recovery route.
class Application(P0FailedLeaseRecoveryTaskApplication):
def __init__(self, config: Any) -> None:
super().__init__(config)
try:
verifier = GrantRootOwnerApprovalVerifier()
except OwnerApprovalVerifierError as exc:
raise TaskProtocolError(str(exc), 503) from exc
self.lineage_supersession = LineageSupersessionEndpoint(
config.db_path, verifier, self.authenticate_control
)
def health(self) -> dict[str, Any]:
value = dict(super().health())
value.update({
"task_lineage_supersession": True,
"task_lineage_supersession_schema": DECISION_SCHEMA,
"task_lineage_supersession_exact_scope": True,
"task_lineage_supersession_exact_cas": True,
"task_lineage_supersession_append_only": True,
"task_lineage_supersession_caller_payload": False,
})
return value
class Handler(P0FailedLeaseRecoveryHandler):
app: Application | None = None
def do_POST(self) -> None:
path = urlparse(self.path).path.rstrip("/") or "/"
if path != SUPERSESSION_ROUTE:
super().do_POST()
return
def action() -> None:
if self.app is None:
raise TaskProtocolError("SERVICE_UNAVAILABLE", 503)
status, result = self.app.lineage_supersession.post(
self._headers(), self._body()
)
if "error" in result:
raise TaskProtocolError(str(result["error"]), status)
self._json(result, status)
self._run(action)
return APIConfig, Application, Handler
def serve(config=None) -> None:
APIConfig, Application, Handler = build_runtime()
config = config or APIConfig.from_env()
application = Application(config)
Handler.app = application
ThreadingHTTPServer((config.host, config.port), Handler).serve_forever()
if __name__ == "__main__":
serve()
@@ -389,6 +389,8 @@ class CanonicalBridgeInitialExecutionResolver:
class InitialExecutionProjectionStore:
"""Apply the exact initial projection through one durable CAS transaction."""
allows_supersession_child = False
def __init__(
self,
db_path: Path,
@@ -406,6 +408,32 @@ class InitialExecutionProjectionStore:
connection.execute("PRAGMA busy_timeout=30000")
return connection
def _validate_projection_entrypoint(
self, connection: sqlite3.Connection, task_id: str
) -> None:
"""Fence formal lineage children from the legacy projection entrypoint.
A supersession child carries additional append-only lineage evidence that
must be committed with its initial projection. The legacy endpoint has
no such contract, so it must not be able to project one accidentally.
"""
tables = {
str(row[0])
for row in connection.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
)
}
if "task_lineage_supersessions" not in tables:
return
lineage = connection.execute(
"SELECT 1 FROM task_lineage_supersessions WHERE child_task_id=?",
(task_id,),
).fetchone()
if lineage is not None and not self.allows_supersession_child:
raise InitialExecutionProjectionError(
"SUPERSESSION_INITIAL_PROJECTION_ENDPOINT_REQUIRED", 409
)
@staticmethod
def _columns(connection: sqlite3.Connection, table: str) -> dict[str, sqlite3.Row]:
return {
@@ -622,6 +650,7 @@ class InitialExecutionProjectionStore:
).fetchone()
if task is None:
raise InitialExecutionProjectionError("TASK_NOT_FOUND", 404)
self._validate_projection_entrypoint(connection, task_id)
if (
str(task["status"]) != request.expected_status
or int(task["revision"]) != request.expected_revision
@@ -771,6 +800,29 @@ class InitialExecutionProjectionStore:
"trace_id": str(task["task_id"]),
}
def _projection_event_extension(
self,
connection: sqlite3.Connection,
task_id: str,
request: InitialExecutionProjectionRequest,
execution: Mapping[str, Any],
applied_revision: int,
) -> Mapping[str, Any]:
"""Return server-derived fields for a specialized projection event."""
return {}
def _insert_projection_extension(
self,
connection: sqlite3.Connection,
task_id: str,
request: InitialExecutionProjectionRequest,
execution: Mapping[str, Any],
event_payload: Mapping[str, Any],
applied_revision: int,
) -> None:
"""Persist specialized evidence in the caller's projection transaction."""
return None
def project(
self,
task_id: str,
@@ -782,6 +834,7 @@ class InitialExecutionProjectionStore:
request_sha = _sha(request.public())
with self._connect() as connection:
self._validate_projection_entrypoint(connection, task_id)
self._ensure_receipt_table(connection)
replay = connection.execute(
"SELECT * FROM task_initial_execution_projections "
@@ -915,6 +968,20 @@ class InitialExecutionProjectionStore:
"main_write": False,
"production_write": False,
}
extension = dict(
self._projection_event_extension(
connection,
task_id,
request,
execution,
applied_revision,
)
)
if set(extension).intersection(event_payload):
raise InitialExecutionProjectionError(
"INITIAL_EXECUTION_PROJECTION_EVENT_EXTENSION_CONFLICT", 500
)
event_payload.update(extension)
self._insert_event(connection, task_id, event_payload)
connection.execute(
"""
@@ -938,6 +1005,14 @@ class InitialExecutionProjectionStore:
_now(),
),
)
self._insert_projection_extension(
connection,
task_id,
request,
execution,
event_payload,
applied_revision,
)
connection.commit()
return {
"task_id": task_id,
+102 -3
View File
@@ -24,7 +24,13 @@ from gitea_tag_publication import (
validate_version,
validate_gitea_api_url,
)
from orchestrator.action_monitor import ActionMonitor
try:
from orchestrator.action_monitor import ActionMonitor
except ModuleNotFoundError:
# The deployed service normally supplies runtime/agent-os on sys.path.
# Keep direct, isolated contract execution bound to the repository copy.
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "runtime" / "agent-os"))
from orchestrator.action_monitor import ActionMonitor
from silent_release_hardening import (
ApprovalKeyRing,
ApprovalRevocationStore,
@@ -64,6 +70,11 @@ except ModuleNotFoundError:
repository_snapshot,
run_authoritative_git,
)
try:
from aios_repository_discovery import RepositoryDiscovery, RepositoryDiscoveryError
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parent))
from aios_repository_discovery import RepositoryDiscovery, RepositoryDiscoveryError
ROOT = Path(os.getenv('AIOS_GITOPS_ROOT', '/data/aios-gitops'))
WS = Path(os.getenv('AIOS_GITOPS_WORKSPACE', str(ROOT / 'workspace')))
@@ -109,6 +120,8 @@ READ_ONLY_GET_PATHS = frozenset({
'/admin/ai-evolution/api/data',
'/api/release-consistency', '/api/tag-publication-status',
'/api/release-acceptance',
'/api/repository/tree', '/api/repository/search',
'/api/repository/diff-index', '/api/repository/file',
'/getReleaseCandidate', '/getReleasePlan', '/getReleaseAcceptance',
'/getApprovalEnvelope', '/getReleaseSaga', '/getCapabilityStatus',
'/getRecoveryStatus',
@@ -342,6 +355,77 @@ def required_authoritative_git_dir() -> Path:
return Path(str(snapshot["authoritative_git_dir"]))
_DISCOVERY_ACTIONS = {
'/api/repository/tree': (
'list_repository_tree',
{'repository', 'ref', 'limit'},
{'prefix', 'depth', 'cursor'},
),
'/api/repository/search': (
'search_repository_source',
{'repository', 'ref', 'query', 'max_matches', 'max_bytes'},
{'path_globs', 'cursor'},
),
'/api/repository/diff-index': (
'get_diff_index',
{'repository', 'from', 'to', 'limit'},
{'path_prefix', 'cursor'},
),
'/api/repository/file': (
'read_repository_file',
{'repository', 'ref', 'path'},
{'max_bytes'},
),
}
def repository_discovery_request(
path: str, query: dict[str, list[str]]
) -> dict[str, Any]:
operation, required, optional = _DISCOVERY_ACTIONS[path]
if set(query) - required - optional:
raise ValueError('unknown repository discovery query field')
request: dict[str, Any] = {}
for key, values in query.items():
if key == 'path_globs':
if not values:
raise ValueError('empty repository discovery path glob')
request[key] = list(values)
continue
if len(values) != 1:
raise ValueError('duplicate repository discovery query field')
value: Any = values[0]
if key in {'depth', 'limit', 'max_matches', 'max_bytes'}:
try:
value = int(value, 10)
except (TypeError, ValueError) as exc:
raise ValueError('invalid repository discovery integer') from exc
request[key] = value
if required - set(request):
raise ValueError('missing repository discovery query field')
if not REPOSITORY_IDENTITY:
raise RepositoryDiscoveryError('REPOSITORY_DISCOVERY_UNAVAILABLE')
try:
discovery = RepositoryDiscovery(
allowed_repositories=(REPOSITORY_IDENTITY,),
git_dir=required_authoritative_git_dir(),
cursor_secret=os.getenv('AIOS_REPOSITORY_DISCOVERY_CURSOR_SECRET'),
)
except RuntimeError as exc:
if isinstance(exc, RepositoryDiscoveryError):
raise
raise RepositoryDiscoveryError('REPOSITORY_DISCOVERY_UNAVAILABLE') from exc
public_to_internal = {
'ref': 'revision',
'path': 'file_path',
'path_prefix': 'prefix',
}
internal_request = {
public_to_internal.get(key, key): value for key, value in request.items()
}
return dict(getattr(discovery, operation)(internal_request))
def publication_publisher() -> TagPublisher:
authoritative_repo = required_authoritative_git_dir()
publisher = TagPublisher(
@@ -441,6 +525,8 @@ class H(BaseHTTPRequestHandler):
return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
def do_POST(self):
path = urllib.parse.urlparse(self.path).path
if path in _DISCOVERY_ACTIONS:
return self.sendj({'error': 'method_not_allowed'},405)
if path in {
'/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
'/buildProductionArtifact', '/executeApprovedProductionRelease',
@@ -456,12 +542,21 @@ class H(BaseHTTPRequestHandler):
def do_GET(self):
u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
q=urllib.parse.parse_qs(u.query)
if len(self.path) > 8192:
return self.sendj({'error':'query_too_large'},413)
try:
q=urllib.parse.parse_qs(
u.query, keep_blank_values=True, max_num_fields=64
)
except ValueError:
return self.sendj({'error':'query_invalid'},400)
try:
if path in ('/health','/api/health'): return self.sendj({'status':'ok','service':'chatgpt-gitops-adapter','workspace':str(WS)})
if path in ('/api/openapi.yaml','/api/openapi-chatgpt-actions.yaml','/api/schema'):
return self.sendt(OPENAPI.read_text(encoding='utf-8'), 'application/yaml; charset=utf-8') if OPENAPI.exists() else self.sendj({'error':'schema_missing'},404)
if path in ('/api','/api/status'): return self.sendj(api_spec())
if path in _DISCOVERY_ACTIONS:
return self.sendj(repository_discovery_request(path, q))
if path in ('/api/base','/api/baseline','/base'):
data=jload(WS/'baseline'/'current.json', {})
data.setdefault('authoritative_repository', repository_snapshot(
@@ -648,6 +743,8 @@ class H(BaseHTTPRequestHandler):
return self.sendj({'error':'not_found','path':path},404)
except FileNotFoundError:
return self.sendj({'error':'release candidate not found'},404)
except RepositoryDiscoveryError as e:
return self.sendj({'error': redact(e)}, getattr(e, 'status_code', 400))
except ValueError as e:
error=redact(e)
monitor.record_failure('do_GET', error, {'path':path})
@@ -664,6 +761,8 @@ class H(BaseHTTPRequestHandler):
return self.sendj({'error':'not_found','path':path},404)
def do_POST(self):
path=urllib.parse.urlparse(self.path).path.rstrip('/')
if path in _DISCOVERY_ACTIONS:
return self.sendj({'error':'method_not_allowed'},405)
if reject_non_read_http(self, 'POST', path): return
data=self.body()
try:
@@ -960,7 +1059,7 @@ return{vs,cmds,ar,cmd,cdet,tot,sok,sfail,runs,baseline,refreshMsg,loadCmd,selVer
def api_spec():
return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],'write':['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl','POST /createReleaseCandidate','POST /submitOwnerApproval','POST /startDeployment'],'policy':{'bearer_required_for_write':True,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True,'silent_release_execution_enabled':SILENT_RELEASE_EXECUTION_ENABLED,'silent_release_shadow_default':True}}
return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/repository/tree','GET /api/repository/search','GET /api/repository/diff-index','GET /api/repository/file','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],'write':['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl','POST /createReleaseCandidate','POST /submitOwnerApproval','POST /startDeployment'],'policy':{'bearer_required_for_write':True,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True,'silent_release_execution_enabled':SILENT_RELEASE_EXECUTION_ENABLED,'silent_release_shadow_default':True}}
def main():
validate_gitea_api_url(os.getenv('AIOS_GITOPS_GITEA_API_URL', ''))
+42 -1
View File
@@ -33,6 +33,10 @@ BRIDGE_OPERATIONS = (
"getReleaseConsistency",
"runReleaseAcceptance",
"getControlPlaneSnapshot",
"listRepositoryTree",
"searchRepositorySource",
"getDiffIndex",
"readRepositoryFile",
)
TASK_ADMISSION_OPERATIONS = tuple(TASK_ADMISSION_OPERATION_REQUIRED_FIELDS)
@@ -55,8 +59,31 @@ OPERATION_REQUIRED_FIELDS = {
"runReleaseAcceptance": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
"getControlPlaneSnapshot": (),
**TASK_ADMISSION_OPERATION_REQUIRED_FIELDS,
"listRepositoryTree": ("repository", "ref", "limit"),
"searchRepositorySource": ("repository", "ref", "query", "max_matches", "max_bytes"),
"getDiffIndex": ("repository", "from", "to", "limit"),
"readRepositoryFile": ("repository", "ref", "path"),
}
OPERATION_OPTIONAL_FIELDS = {
"listRepositoryTree": ("prefix", "depth", "cursor"),
"searchRepositorySource": ("path_globs", "cursor"),
"getDiffIndex": ("path_prefix", "cursor"),
"readRepositoryFile": ("max_bytes",),
}
READ_ONLY_OPERATIONS = frozenset(
{
"getRuntimeStatus",
"getGovernedWriteCapabilities",
"getControlPlaneSnapshot",
"listRepositoryTree",
"searchRepositorySource",
"getDiffIndex",
"readRepositoryFile",
}
)
_OPERATION_PATHS = {
"createGovernedTask": "/v2/governed-task",
"projectOnlineAgentExecution": "/v2/execution-envelope",
@@ -70,6 +97,10 @@ _OPERATION_PATHS = {
"getReleaseConsistency": "/v2/release/consistency",
"runReleaseAcceptance": "/v2/release/acceptance",
**{operation: path for path, operation in TASK_ADMISSION_ROUTES.items()},
"listRepositoryTree": "/v2/repository/tree",
"searchRepositorySource": "/v2/repository/search",
"getDiffIndex": "/v2/repository/diff-index",
"readRepositoryFile": "/v2/repository/file",
}
@@ -87,9 +118,11 @@ def operation_contracts(
"""Return the runtime contract consumed by adapters and schema tooling."""
contracts = {
operation: {
"method": "POST",
"method": "GET" if operation in OPERATION_OPTIONAL_FIELDS else "POST",
"path": _OPERATION_PATHS.get(operation),
"required_fields": list(fields),
"optional_fields": list(OPERATION_OPTIONAL_FIELDS.get(operation, ())),
"read_only": operation in READ_ONLY_OPERATIONS,
}
for operation, fields in OPERATION_REQUIRED_FIELDS.items()
if operation not in TASK_ADMISSION_OPERATIONS
@@ -281,6 +314,10 @@ class BridgeV2Adapter:
required = OPERATION_REQUIRED_FIELDS[operation]
if any(key not in request for key in required):
raise BridgeV2Error("BRIDGE_REQUEST_FIELDS_MISSING")
if operation in OPERATION_OPTIONAL_FIELDS:
allowed = set(required) | set(OPERATION_OPTIONAL_FIELDS[operation])
if set(request) - allowed:
raise BridgeV2Error("BRIDGE_REQUEST_FIELDS_UNKNOWN")
handler = self.handlers.get(operation)
if handler is None:
raise BridgeV2Error("GOVERNED_HANDLER_UNAVAILABLE")
@@ -304,5 +341,9 @@ class BridgeV2Adapter:
def getTaskAdmissionEnvelope(self, request: Mapping[str, Any]) -> Any: return self._call("getTaskAdmissionEnvelope", request)
def submitTaskAdmissionAuthorization(self, request: Mapping[str, Any]) -> Any: return self._call("submitTaskAdmissionAuthorization", request)
def getTaskAdmissionStatus(self, request: Mapping[str, Any]) -> Any: return self._call("getTaskAdmissionStatus", request)
def listRepositoryTree(self, request: Mapping[str, Any]) -> Any: return self._call("listRepositoryTree", request)
def searchRepositorySource(self, request: Mapping[str, Any]) -> Any: return self._call("searchRepositorySource", request)
def getDiffIndex(self, request: Mapping[str, Any]) -> Any: return self._call("getDiffIndex", request)
def readRepositoryFile(self, request: Mapping[str, Any]) -> Any: return self._call("readRepositoryFile", request)
def getControlPlaneSnapshot(self, request: Mapping[str, Any] | None = None) -> Any: return self._call("getControlPlaneSnapshot", request or {})
@@ -11,6 +11,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import re
import sqlite3
import stat
from typing import Any, Callable, Mapping
from urllib.parse import parse_qsl, urlsplit
@@ -18,11 +19,13 @@ from aios_authority_git_executor import AuthorityGitExecutor
from aios_chatgpt_bridge_v2 import (
BridgeV2Adapter,
BridgeV2Error,
OPERATION_OPTIONAL_FIELDS,
OPERATION_REQUIRED_FIELDS,
TASK_ADMISSION_OPERATIONS,
TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS,
TASK_ADMISSION_ROUTES,
)
from aios_repository_discovery import RepositoryDiscovery, RepositoryDiscoveryError
from aios_task_admission_bootstrap import TaskAdmissionError
from gitea_tag_publication import validate_gitea_api_url
from aios_continuation_executor import AuthorityGitAdapter, ContinuationCommit, ContinuationExecutor, ContinuationPermissions, ContinuationRuntimeBinding, DetachedGitWorkspace, GiteaContinuationRemote, ServerSideCredentialProjection
@@ -59,7 +62,13 @@ ROUTES = {
"/v2/release/execute": "executeApprovedProductionRelease",
"/v2/release/consistency": "getReleaseConsistency",
"/v2/release/acceptance": "runReleaseAcceptance",
**TASK_ADMISSION_ROUTES,
"/v2/task-admission/envelope": "getTaskAdmissionEnvelope",
"/v2/task-admission/authorize": "submitTaskAdmissionAuthorization",
"/v2/task-admission/status": "getTaskAdmissionStatus",
"/v2/repository/tree": "listRepositoryTree",
"/v2/repository/search": "searchRepositorySource",
"/v2/repository/diff-index": "getDiffIndex",
"/v2/repository/file": "readRepositoryFile",
}
# Admission is deliberately separate from a TaskPrincipal. A successful
@@ -93,6 +102,7 @@ class RuntimeConfig:
baseline_path: Path = Path("/var/lib/aios-gitops-runtime-anchor/baseline/current.json")
health_url: str = "http://127.0.0.1:18927/health"
release_plan_path: Path | None = None
repository_discovery_cursor_secret: str | bytes | None = None
def _safe(value: str) -> str:
@@ -109,6 +119,7 @@ class GovernedRuntime:
self.controllers: dict[str, GitOpsJobController] = {}
self.bindings: dict[str, dict[str, Any]] = {}
self.release = build_executor_config(config)
self._repository_discovery_instance: RepositoryDiscovery | None = None
metadata_root = config.state_root / "task-bindings"
if metadata_root.is_dir():
for metadata in metadata_root.glob("*.json"):
@@ -324,6 +335,32 @@ class GovernedRuntime:
)
result = self.bootstrap.create(request); self.bindings[result["task_id"]] = result; return result
def _repository_discovery(self) -> RepositoryDiscovery:
if self._repository_discovery_instance is None:
self._repository_discovery_instance = RepositoryDiscovery(
allowed_repositories=(self.config.repository,),
git_dir=self.config.authoritative_git_dir,
cursor_secret=(
self.config.repository_discovery_cursor_secret
if self.config.repository_discovery_cursor_secret is not None
else os.environ.get("AIOS_REPOSITORY_DISCOVERY_CURSOR_SECRET")
),
)
return self._repository_discovery_instance
def repository_discovery(
self, operation: str, request: Mapping[str, Any]
) -> dict[str, Any]:
method = {
"listRepositoryTree": "list_repository_tree",
"searchRepositorySource": "search_repository_source",
"getDiffIndex": "get_diff_index",
"readRepositoryFile": "read_repository_file",
}.get(operation)
if method is None:
raise ValueError("REPOSITORY_DISCOVERY_OPERATION_INVALID")
return getattr(self._repository_discovery(), method)(request)
def acquire_lease(self, request: Mapping[str, Any]) -> dict[str, Any]:
replay = self._durable_completed_replay_lease(request)
if replay is not None:
@@ -895,6 +932,18 @@ class GovernedRuntime:
"resumeGitOpsJob": lambda request: self.job(request, "resume"),
"reconcileGitOpsJob": lambda request: self.job(request, "reconcile"),
"getControlPlaneSnapshot": lambda request: self.snapshot(),
"listRepositoryTree": lambda request: self.repository_discovery(
"listRepositoryTree", request
),
"searchRepositorySource": lambda request: self.repository_discovery(
"searchRepositorySource", request
),
"getDiffIndex": lambda request: self.repository_discovery(
"getDiffIndex", request
),
"readRepositoryFile": lambda request: self.repository_discovery(
"readRepositoryFile", request
),
"createTagApprovalRequest": self.release.approval_request,
"requestOrCreateTag": self.release.publish_tag,
"getTagPublicationStatus": self.release.tag_status,
@@ -969,11 +1018,15 @@ class BridgeHandler(BaseHTTPRequestHandler):
allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | set(
TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS[operation]
)
elif operation in OPERATION_OPTIONAL_FIELDS:
allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | set(
OPERATION_OPTIONAL_FIELDS[operation]
)
else:
allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {"schema", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required", "expected_branch_oid", "expected_revision", "generation", "branch_mode", "source_lease_epoch"}
if operation == "projectOnlineAgentExecution": allowed.add("execution")
if operation == "commitFilesGoverned": allowed.add("files")
if operation not in TASK_ADMISSION_OPERATIONS:
if operation not in TASK_ADMISSION_OPERATIONS and operation not in OPERATION_OPTIONAL_FIELDS:
allowed.update({"owner_approval", "execute", "job_id", "historical_lease_head_oid"})
if "principal" in body or _PRIVATE_PRINCIPAL in body:
raise _HTTPError("CALLER_PRINCIPAL_FORBIDDEN")
@@ -1024,12 +1077,47 @@ class BridgeHandler(BaseHTTPRequestHandler):
raise _HTTPError("TASK_PRINCIPAL_REJECTED")
return {**body, _PRIVATE_PRINCIPAL: dict(principal)}
def _query_body(self, operation: str) -> dict[str, Any]:
parsed = urlsplit(self.path)
if len(self.path) > 8192:
raise _HTTPError("QUERY_TOO_LARGE")
try:
pairs = parse_qsl(
parsed.query,
keep_blank_values=True,
strict_parsing=False,
max_num_fields=64,
)
except ValueError as exc:
raise _HTTPError("QUERY_INVALID") from exc
allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | set(
OPERATION_OPTIONAL_FIELDS[operation]
)
result: dict[str, Any] = {}
for key, value in pairs:
if key not in allowed:
raise _HTTPError("UNKNOWN_FIELDS")
if key == "path_globs":
result.setdefault(key, []).append(value)
continue
if key in result:
raise _HTTPError("DUPLICATE_FIELD")
if key in {"depth", "limit", "max_matches", "max_bytes"}:
try:
value = int(value, 10)
except (TypeError, ValueError) as exc:
raise _HTTPError("INVALID_INTEGER") from exc
result[key] = value
if set(OPERATION_REQUIRED_FIELDS[operation]) - set(result):
raise _HTTPError("MISSING_FIELDS")
return result
def do_GET(self) -> None:
if self.adapter is None: self._json({"error": "RUNTIME_UNAVAILABLE"}, 503); return
if self.path == "/v2/runtime": self._json(self.adapter.getRuntimeStatus()); return
if self.path == "/v2/capabilities": self._json(self.adapter.getGovernedWriteCapabilities()); return
if self.path == "/v2/control-plane-snapshot": self._json(self.adapter.getControlPlaneSnapshot({})); return
parsed = urlsplit(self.path)
if parsed.path == "/v2/runtime": self._json(self.adapter.getRuntimeStatus()); return
if parsed.path == "/v2/capabilities": self._json(self.adapter.getGovernedWriteCapabilities()); return
if parsed.path == "/v2/control-plane-snapshot": self._json(self.adapter.getControlPlaneSnapshot({})); return
if parsed.path == "/v2/task-admission/status":
if not callable(self.adapter.handlers.get("getTaskAdmissionStatus")):
self._json({"error": "NOT_FOUND"}, 404)
@@ -1055,10 +1143,30 @@ class BridgeHandler(BaseHTTPRequestHandler):
except Exception as exc:
self._json({"error": str(exc) or type(exc).__name__}, 503)
return
operation = ROUTES.get(parsed.path)
if operation in OPERATION_OPTIONAL_FIELDS:
try:
body = self._query_body(operation)
self._allowed(operation, body)
result = getattr(self.adapter, operation)(body)
self._json(result, self._result_status(result))
except _HTTPError as exc:
status = 413 if str(exc) == "QUERY_TOO_LARGE" else 400
self._json({"error": str(exc)}, status)
except RepositoryDiscoveryError as exc:
self._json({"error": exc.code}, exc.status_code)
except (BridgeV2Error, ValueError) as exc:
self._json({"error": str(exc)}, 400)
except Exception as exc:
self._json({"error": str(exc) or type(exc).__name__}, 503)
return
self._json({"error": "NOT_FOUND"}, 404)
def do_POST(self) -> None:
operation = ROUTES.get(self.path)
operation = ROUTES.get(urlsplit(self.path).path)
if operation in OPERATION_OPTIONAL_FIELDS:
self._json({"error": "METHOD_NOT_ALLOWED"}, 405)
return
if operation is None or operation in ("getRuntimeStatus", "getGovernedWriteCapabilities") or self.adapter is None: self._json({"error": "NOT_FOUND"}, 404); return
if operation in TASK_ADMISSION_OPERATIONS and not callable(self.adapter.handlers.get(operation)):
self._json({"error": "NOT_FOUND"}, 404)
@@ -1101,6 +1209,33 @@ def main(
"AIOS_TASK_ADMISSION_ACTIVATION_FILE",
"/var/lib/aios-gitops/bridge-v2-staging/task-admission-activation.json",
).strip()
discovery_secret_file = os.environ.get(
"AIOS_REPOSITORY_DISCOVERY_CURSOR_SECRET_FILE", ""
).strip()
if discovery_secret_file:
secret_path = Path(discovery_secret_file)
try:
secret_metadata = secret_path.lstat()
if (
stat.S_ISLNK(secret_metadata.st_mode)
or not stat.S_ISREG(secret_metadata.st_mode)
or secret_metadata.st_mode & 0o022
or secret_metadata.st_size > 4096
):
raise RuntimeError("REPOSITORY_DISCOVERY_CURSOR_SECRET_INVALID")
discovery_cursor_secret: str | bytes | None = secret_path.read_bytes()
except (OSError, RuntimeError) as exc:
if str(exc) == "REPOSITORY_DISCOVERY_CURSOR_SECRET_INVALID":
raise
raise RuntimeError(
"REPOSITORY_DISCOVERY_CURSOR_SECRET_UNAVAILABLE"
) from exc
if not discovery_cursor_secret:
raise RuntimeError("REPOSITORY_DISCOVERY_CURSOR_SECRET_EMPTY")
else:
discovery_cursor_secret = os.environ.get(
"AIOS_REPOSITORY_DISCOVERY_CURSOR_SECRET"
)
state_root = Path("/var/lib/aios-gitops/bridge-v2-staging")
token_path = Path(os.environ.get("AIOS_GITEA_TOKEN_FILE", "/run/credentials/aios-chatgpt-bridge-v2-staging.service/gitea_token"))
ssh_key_path = Path(os.environ.get("AIOS_GITEA_SSH_KEY_PATH", "/home/aiman/.ssh/id_ed25519_codex_tmp"))
@@ -1120,6 +1255,7 @@ def main(
"AIOS_GITOPS_BASELINE_AUTHORITY",
"/var/lib/aios-gitops-runtime-anchor/baseline/current.json",
)),
repository_discovery_cursor_secret=discovery_cursor_secret,
)
runtime = GovernedRuntime(config)
# A TaskPrincipal verifier is deliberately supplied only by the hosting
@@ -0,0 +1,32 @@
"""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}"
@@ -4,13 +4,22 @@ from __future__ import annotations
from dataclasses import dataclass, asdict
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import tempfile
from typing import Any, Mapping
from aios_gitops_job_controller import GitOpsJobController, JobStore
from aios_gitops_job_identity import legacy_task_storage_key, task_storage_key
from aios_online_agent_execution import GovernedExecutionBinding, project_execution_envelope
try: # pragma: no cover - exercised on the Linux activation host
import fcntl
except ImportError: # pragma: no cover - non-POSIX execution stays fail closed
fcntl = None # type: ignore[assignment]
OID_RE = re.compile(r"^[0-9a-f]{40}$")
BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
@@ -19,6 +28,29 @@ class BootstrapError(RuntimeError):
pass
def _lexists(path: Path) -> bool:
try:
path.lstat()
return True
except FileNotFoundError:
return False
def _require_runtime_directory(path: Path) -> None:
try:
metadata = path.lstat()
except OSError as exc:
raise BootstrapError("TASK_STORAGE_PARENT_INVALID") from exc
expected_uid = getattr(os, "geteuid", lambda: metadata.st_uid)()
if (
stat.S_ISLNK(metadata.st_mode)
or not stat.S_ISDIR(metadata.st_mode)
or metadata.st_mode & 0o022
or metadata.st_uid != expected_uid
):
raise BootstrapError("TASK_STORAGE_PARENT_INVALID")
@dataclass(frozen=True, slots=True)
class GovernedTaskBinding:
task_id: str
@@ -37,25 +69,95 @@ class GovernedTaskBinding:
return {"task_id": self.task_id, "repository": self.repository, "base_oid": self.base_oid, "target_branch": self.target_branch, "expected_branch_oid": self.expected_branch_oid, "generation": self.generation, "branch_mode": self.branch_mode, "create_target_branch": self.branch_mode == "CREATE_NEW", "worktree_mode": "detached_at_base_oid" if self.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True, "idempotent": idempotent, "state": "BOUND", "job_id": self.job_id}
def _safe(value: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "-", value).strip("-")[:120] or "task"
# ===== AIOS Formal Lineage producer-native reservation v12 =====
import re as _fl_re
_FL_PRODUCER_KEY = _fl_re.compile(r"^[0-9a-f]{64}$")
def _formal_lineage_reservation_receipt(
value,
*,
request_idempotency_key: str,
expected_side_effect_state: str,
):
if callable(getattr(value, "as_dict", None)):
raw=value.as_dict()
elif isinstance(value, dict):
raw=dict(value)
else:
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_TYPE_INVALID")
producer_key=raw.get("idempotency_key")
job_id=raw.get("job_id")
if not isinstance(job_id,str) or not job_id:
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_JOB_ID_INVALID")
if raw.get("state")!="REQUESTED":
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_STATE_INVALID")
if raw.get("side_effect_state")!=expected_side_effect_state:
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_SIDE_EFFECT_STATE_INVALID")
if not isinstance(producer_key,str) or not _FL_PRODUCER_KEY.fullmatch(producer_key):
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_PRODUCER_IDEMPOTENCY_INVALID")
if job_id!="job-"+producer_key[:32]:
raise RuntimeError("GOVERNED_TASK_SUBMIT_RECEIPT_JOB_PRODUCER_KEY_CONFLICT")
if not isinstance(request_idempotency_key,str) or not request_idempotency_key:
raise RuntimeError("GOVERNED_TASK_REQUEST_IDEMPOTENCY_INVALID")
return {
"job_id":job_id,
"state":"REQUESTED",
"side_effect_state":raw["side_effect_state"],
"producer_idempotency_key":producer_key,
"request_idempotency_key":request_idempotency_key,
}
def _formal_lineage_attach_receipt(response, receipt):
if not isinstance(response, dict):
raise RuntimeError("GOVERNED_TASK_BINDING_RESPONSE_INVALID")
if response.get("state")!="BOUND":
raise RuntimeError("GOVERNED_TASK_BINDING_STATE_INVALID")
if response.get("job_id")!=receipt.get("job_id"):
raise RuntimeError("GOVERNED_TASK_BINDING_RECEIPT_JOB_CONFLICT")
result=dict(response)
result["reservation_receipt"]=dict(receipt)
return result
class GovernedTaskBootstrap:
def __init__(self, authority: Any, root: Path, *, remote_name: str, token_path: Path, ssh_key_path: Path, api_base: str = "https://g.cnnetmkt.top/git/api/v1") -> None:
self.authority = authority; self.root = Path(root); self.remote_name = remote_name; self.token_path = Path(token_path); self.ssh_key_path = Path(ssh_key_path); self.api_base = api_base; self.task_meta_root = self.root / "task-bindings"; self.job_root = self.root / "jobs"; self.task_meta_root.mkdir(parents=True, exist_ok=True); self.job_root.mkdir(parents=True, exist_ok=True)
self.authority = authority; self.root = Path(root); self.remote_name = remote_name; self.token_path = Path(token_path); self.ssh_key_path = Path(ssh_key_path); self.api_base = api_base; self.task_meta_root = self.root / "task-bindings"; self.job_root = self.root / "jobs"; self.lock_root = self.root / "task-binding-locks"; self.task_meta_root.mkdir(parents=True, exist_ok=True, mode=0o700); self.job_root.mkdir(parents=True, exist_ok=True, mode=0o700); self.lock_root.mkdir(parents=True, exist_ok=True, mode=0o700); _require_runtime_directory(self.task_meta_root); _require_runtime_directory(self.job_root); _require_runtime_directory(self.lock_root)
def _store(self, task_id: str) -> tuple[JobStore, GitOpsJobController]:
root = self.job_root / _safe(task_id); store = JobStore(root, payload_root=root / "payloads"); return store, GitOpsJobController(store)
root = self.job_root / task_storage_key(task_id)
legacy = self.job_root / legacy_task_storage_key(task_id)
if legacy != root and _lexists(legacy):
if _lexists(root):
raise BootstrapError("TASK_STORAGE_LEGACY_CONFLICT")
raise BootstrapError("TASK_STORAGE_LEGACY_MIGRATION_REQUIRED")
store = JobStore(root, payload_root=root / "payloads")
return store, GitOpsJobController(store)
def _metadata(self, task_id: str) -> Path:
return self.task_meta_root / (_safe(task_id) + ".json")
return self.task_meta_root / (task_storage_key(task_id) + ".json")
def _legacy_metadata(self, task_id: str) -> Path:
return self.task_meta_root / (legacy_task_storage_key(task_id) + ".json")
def _read_existing(self, task_id: str) -> GovernedTaskBinding | None:
path = self._metadata(task_id)
if not path.is_file() or path.is_symlink(): return None
legacy = self._legacy_metadata(task_id)
if legacy != path and _lexists(legacy):
if _lexists(path):
raise BootstrapError("TASK_BINDING_LEGACY_CONFLICT")
raise BootstrapError("TASK_BINDING_LEGACY_MIGRATION_REQUIRED")
if not _lexists(path):
return None
if not path.is_file() or path.is_symlink():
raise BootstrapError("TASK_BINDING_PATH_INVALID")
return GovernedTaskBinding(**json.loads(path.read_text(encoding="utf-8")))
binding = GovernedTaskBinding(**json.loads(path.read_text(encoding="utf-8")))
if binding.task_id != task_id:
raise BootstrapError("TASK_BINDING_IDENTITY_CONFLICT")
return binding
def get_binding(self, task_id: str) -> GovernedTaskBinding:
"""Read the exact durable binding; never synthesize missing lineage."""
binding = self._read_existing(str(task_id))
@@ -96,29 +198,106 @@ class GovernedTaskBootstrap:
)
def _write(self, binding: GovernedTaskBinding) -> None:
path = self._metadata(binding.task_id); temporary = path.with_suffix(".tmp"); temporary.write_text(json.dumps(asdict(binding), sort_keys=True), encoding="utf-8"); temporary.replace(path)
path = self._metadata(binding.task_id)
descriptor, temporary_name = tempfile.mkstemp(
prefix=".task-binding-", dir=self.task_meta_root,
)
temporary = Path(temporary_name)
try:
os.fchmod(descriptor, 0o600)
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
stream.write(json.dumps(asdict(binding), sort_keys=True) + "\n")
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
os.chmod(path, 0o600)
parent = os.open(self.task_meta_root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(parent)
finally:
os.close(parent)
except BaseException:
temporary.unlink(missing_ok=True)
raise
def _task_lock(self, task_id: str) -> int:
if fcntl is None:
raise BootstrapError("TASK_BINDING_LOCK_UNAVAILABLE")
path = self.lock_root / (task_storage_key(task_id) + ".lock")
descriptor = os.open(
path,
os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0),
0o600,
)
metadata = os.fstat(descriptor)
expected_uid = getattr(os, "geteuid", lambda: metadata.st_uid)()
if (
not stat.S_ISREG(metadata.st_mode)
or metadata.st_nlink != 1
or metadata.st_uid != expected_uid
):
os.close(descriptor)
raise BootstrapError("TASK_BINDING_LOCK_INVALID")
os.fchmod(descriptor, 0o600)
try:
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError as exc:
os.close(descriptor)
raise BootstrapError("TASK_BINDING_BUSY") from exc
return descriptor
@staticmethod
def _unlock_task(descriptor: int) -> None:
if fcntl is not None:
fcntl.flock(descriptor, fcntl.LOCK_UN)
os.close(descriptor)
def create(self, request: Mapping[str, Any]) -> dict[str, Any]:
values = {key: str(request[key]) for key in ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")}
if values["repository"] != "aios/aios" or not OID_RE.fullmatch(values["base_oid"]): raise BootstrapError("TASK_BINDING_INVALID")
if not BRANCH_RE.fullmatch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
existing = self._read_existing(values["task_id"])
if existing is not None:
if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description")): raise BootstrapError("TASK_BINDING_CONFLICT")
return existing.response(idempotent=True)
if not self.authority.commit_exists(values["base_oid"]): raise BootstrapError("BASE_COMMIT_NOT_FOUND")
current = self.authority.read_ref("refs/heads/" + values["target_branch"])
expected = str(request.get("expected_branch_oid") or values["base_oid"])
if current is None:
mode, generation = "CREATE_NEW", 1
if expected != values["base_oid"]: raise BootstrapError("EXPECTED_BASE_CONFLICT")
else:
if expected != current: raise BootstrapError("EXISTING_BRANCH_OID_CONFLICT")
mode, generation = "CONTINUE_EXISTING", int(request.get("generation", 2))
if generation < 1: raise BootstrapError("GENERATION_INVALID")
payload = json.dumps(values, sort_keys=True, separators=(",", ":")).encode()
_, controller = self._store(values["task_id"])
receipt = controller.submit({"idempotency_key": values["idempotency_key"], "campaign_id": values["task_id"], "task_id": values["task_id"], "repository": values["repository"], "operation": "GOVERNED_TASK_CREATE", "branch": values["target_branch"], "expected_branch_oid": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)
binding = GovernedTaskBinding(**values, expected_branch_oid=expected, branch_mode=mode, generation=generation, job_id=receipt.job_id)
self._write(binding)
return binding.response()
descriptor = self._task_lock(values["task_id"])
try:
existing = self._read_existing(values["task_id"])
if existing is not None:
if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")): raise BootstrapError("TASK_BINDING_CONFLICT")
expected = str(request.get('expected_branch_oid') or values['base_oid'])
payload = json.dumps(values, sort_keys=True, separators=(',', ':')).encode()
_, controller = self._store(values['task_id'])
_fl_submit_receipt = controller.submit({'idempotency_key': values['idempotency_key'], 'campaign_id': values['task_id'], 'task_id': values['task_id'], 'repository': values['repository'], 'operation': 'GOVERNED_TASK_CREATE', 'branch': values['target_branch'], 'expected_branch_oid': expected, 'payload_digest': hashlib.sha256(payload).hexdigest(), 'lineage_anchor_oid': values['base_oid']}, payload=payload)
_fl_receipt = _formal_lineage_reservation_receipt(
_fl_submit_receipt, request_idempotency_key=request["idempotency_key"],
expected_side_effect_state='SAME_JOB'
)
_fl_response = existing.response(idempotent=True)
return _formal_lineage_attach_receipt(_fl_response,_fl_receipt)
if not self.authority.commit_exists(values["base_oid"]): raise BootstrapError("BASE_COMMIT_NOT_FOUND")
current = self.authority.read_ref("refs/heads/" + values["target_branch"])
expected = str(request.get("expected_branch_oid") or values["base_oid"])
if current is None:
mode, generation = "CREATE_NEW", 1
if expected != values["base_oid"]: raise BootstrapError("EXPECTED_BASE_CONFLICT")
else:
if expected != current: raise BootstrapError("EXISTING_BRANCH_OID_CONFLICT")
mode, generation = "CONTINUE_EXISTING", int(request.get("generation", 2))
if generation < 1: raise BootstrapError("GENERATION_INVALID")
payload = json.dumps(values, sort_keys=True, separators=(",", ":")).encode()
_, controller = self._store(values["task_id"])
expected_job_id = controller.internal_job_id_for_idempotency_key(values["idempotency_key"])
with controller.store._connect() as connection:
bootstrap_jobs = connection.execute(
"SELECT job_id FROM jobs WHERE operation='GOVERNED_TASK_CREATE' ORDER BY job_id"
).fetchall()
if bootstrap_jobs and [str(row[0]) for row in bootstrap_jobs] != [expected_job_id]:
raise BootstrapError("TASK_BINDING_RECOVERY_CONFLICT")
receipt = controller.submit({"idempotency_key": values["idempotency_key"], "campaign_id": values["task_id"], "task_id": values["task_id"], "repository": values["repository"], "operation": "GOVERNED_TASK_CREATE", "branch": values["target_branch"], "expected_branch_oid": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)
binding = GovernedTaskBinding(**values, expected_branch_oid=expected, branch_mode=mode, generation=generation, job_id=receipt.job_id)
self._write(binding)
_fl_receipt = _formal_lineage_reservation_receipt(
receipt, request_idempotency_key=request["idempotency_key"],
expected_side_effect_state="QUEUED"
)
_fl_response = binding.response()
return _formal_lineage_attach_receipt(_fl_response,_fl_receipt)
finally:
self._unlock_task(descriptor)
@@ -0,0 +1,967 @@
"""Bounded, read-only discovery over a verified immutable Git repository.
The module is deliberately independent from Task, lease, credential, worktree,
and write code. Every public operation verifies the exact repository identity,
resolves a floating ref once to a full commit OID, and then reads only that
immutable snapshot.
"""
from __future__ import annotations
import base64
from collections.abc import Iterable, Iterator, Mapping
import fnmatch
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import subprocess
import time
from authoritative_git import AuthoritativeGitError, verify_repository
_OID_RE = re.compile(r"^[0-9a-f]{40}$")
_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,239}$")
_IDENTITY_RE = re.compile(
r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}/[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)
_READ_ONLY_COMMANDS = frozenset({"cat-file", "diff", "ls-tree", "rev-parse"})
_MAX_COMMAND_OUTPUT = 1_048_576
_MAX_RECORD_BYTES = 1_048_576
_MAX_LIMIT = 1_000
_MAX_DEPTH = 32
_MAX_BYTES = 1_048_576
_MAX_GLOBS = 16
_MAX_QUERY_LENGTH = 512
_MAX_CURSOR_LENGTH = 4_096
_GIT_TIMEOUT_SECONDS = 8.0
class RepositoryDiscoveryError(RuntimeError):
"""A fail-closed repository discovery request error."""
def __init__(self, code: str, *, status_code: int | None = None) -> None:
self.code = str(code)
self.status_code = (
status_code if status_code is not None else self._status_for(self.code)
)
super().__init__(self.code)
@staticmethod
def _status_for(code: str) -> int:
if "OUTPUT_LIMIT" in code:
return 413
if code.endswith("_NOT_FOUND"):
return 404
if code in {"REPOSITORY_NOT_ALLOWED", "REPOSITORY_IDENTITY_MISMATCH"}:
return 403
if code in {
"REPOSITORY_DISCOVERY_UNAVAILABLE",
"REPOSITORY_DISCOVERY_GIT_FAILED",
"REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID",
"REPOSITORY_DISCOVERY_CURSOR_SECRET_REQUIRED",
}:
return 503
return 400
class RepositoryDiscoveryNotFound(RepositoryDiscoveryError):
"""A validated immutable object or path was not found."""
class RepositoryDiscoveryOutputTooLarge(RepositoryDiscoveryError):
"""A fixed read-only Git query exceeded its collection boundary."""
def _error(code: str) -> RepositoryDiscoveryError:
return RepositoryDiscoveryError(code)
def _request_mapping(request: Mapping[str, object]) -> Mapping[str, object]:
if not isinstance(request, Mapping):
raise _error("REPOSITORY_DISCOVERY_REQUEST_INVALID")
return request
def _required_request_value(request: Mapping[str, object], field: str) -> object:
value = request.get(field)
if value is None:
raise _error("REPOSITORY_DISCOVERY_REQUEST_INVALID")
return value
def _canonical(value: Mapping[str, object]) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
).encode("ascii")
def _positive_int(value: object, field: str, maximum: int) -> int:
if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= maximum:
raise _error(f"REPOSITORY_DISCOVERY_INVALID_{field.upper()}")
return value
def _safe_relative_path(
value: object,
field: str,
*,
allow_empty: bool,
allow_glob: bool = False,
) -> str:
if not isinstance(value, str) or value != value.strip() or "\x00" in value:
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
if "\\" in value:
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
if value in ("", "."):
if allow_empty:
return ""
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
if value.startswith(("/", "-")) or "//" in value or ":" in value:
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
if not allow_glob and any(character in value for character in "*?["):
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
parts = value.split("/")
if any(part in ("", ".", "..") for part in parts):
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
if len(value) > 512:
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
return value
def _safe_glob(value: object) -> str:
return _safe_relative_path(value, "path_glob", allow_empty=False, allow_glob=True)
def _safe_ref(value: str) -> bool:
return bool(
_REF_RE.fullmatch(value)
and ".." not in value
and "//" not in value
and "@{" not in value
and not value.endswith(("/", ".", ".lock"))
and not any(part in ("", ".", "..") for part in value.split("/"))
)
def _parse_tree_record(record: bytes) -> dict[str, str] | None:
try:
header, raw_path = record.split(b"\t", 1)
mode, kind, oid = header.decode("ascii").split(" ", 2)
path = raw_path.decode("utf-8")
except (UnicodeDecodeError, ValueError) as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
if kind not in {"blob", "tree", "commit"} or not _OID_RE.fullmatch(oid.lower()):
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
try:
_safe_relative_path(path, "tree_path", allow_empty=False)
except RepositoryDiscoveryError:
# Git permits filenames outside the public POSIX-relative contract.
# They are not exposed as usable discovery paths.
return None
return {"mode": mode, "type": kind, "object_oid": oid.lower(), "path": path}
def _decode_output_path(raw: bytes) -> str:
try:
value = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
return _safe_relative_path(value, "git_output_path", allow_empty=False)
class RepositoryDiscovery:
"""Repository-scoped immutable Git discovery with signed cursors."""
def __init__(
self,
*,
allowed_repositories: Iterable[str] | None = None,
git_dir: str | Path | None = None,
cursor_secret: str | bytes | None = None,
) -> None:
configured = allowed_repositories
if configured is None:
configured = (os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""),)
identities = tuple(configured)
if not identities or any(
not isinstance(item, str) or not _IDENTITY_RE.fullmatch(item)
for item in identities
):
raise _error("REPOSITORY_DISCOVERY_ALLOWLIST_INVALID")
self._allowed_repositories = frozenset(identities)
self._git_dir = git_dir
secret = (
cursor_secret
if cursor_secret is not None
else os.getenv("AIOS_REPOSITORY_DISCOVERY_CURSOR_SECRET", "")
)
if isinstance(secret, str):
secret = secret.encode("utf-8")
if not isinstance(secret, bytes) or len(secret) < 16:
raise _error("REPOSITORY_DISCOVERY_CURSOR_SECRET_REQUIRED")
self._cursor_secret = secret
def _repository_path(self, repository: object) -> Path:
if not isinstance(repository, str) or repository not in self._allowed_repositories:
raise _error("REPOSITORY_NOT_ALLOWED")
try:
path, actual_identity = verify_repository(
git_dir=self._git_dir,
repository_identity=repository,
)
except AuthoritativeGitError as exc:
code = (
"REPOSITORY_IDENTITY_MISMATCH"
if "REPOSITORY_IDENTITY_MISMATCH" in str(exc)
else "REPOSITORY_DISCOVERY_UNAVAILABLE"
)
raise _error(code) from exc
if actual_identity != repository:
raise _error("REPOSITORY_IDENTITY_MISMATCH")
return path
@staticmethod
def _git_env() -> dict[str, str]:
return {
**os.environ,
"GIT_OPTIONAL_LOCKS": "0",
"GIT_CONFIG_NOSYSTEM": "1",
"GIT_TERMINAL_PROMPT": "0",
}
@staticmethod
def _run(
path: Path,
args: list[str],
*,
max_output: int = _MAX_COMMAND_OUTPUT,
) -> bytes:
if not args or args[0] not in _READ_ONLY_COMMANDS:
raise _error("REPOSITORY_DISCOVERY_GIT_COMMAND_REJECTED")
output_limit = min(max_output, _MAX_COMMAND_OUTPUT)
process: subprocess.Popen[bytes] | None = None
try:
process = subprocess.Popen(
["git", "--git-dir", str(path), *args],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
env=RepositoryDiscovery._git_env(),
)
assert process.stdout is not None
output = process.stdout.read(output_limit + 1)
if len(output) > output_limit:
process.kill()
process.wait(timeout=1)
raise RepositoryDiscoveryOutputTooLarge(
"REPOSITORY_DISCOVERY_OUTPUT_LIMIT"
)
returncode = process.wait(timeout=_GIT_TIMEOUT_SECONDS)
except RepositoryDiscoveryError:
raise
except (OSError, subprocess.TimeoutExpired) as exc:
if process is not None:
try:
process.kill()
process.wait(timeout=1)
except (OSError, subprocess.TimeoutExpired):
pass
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED") from exc
if returncode != 0:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED")
return output
@staticmethod
def _read_blob_prefix(path: Path, oid: str, maximum: int) -> tuple[bytes, bool]:
"""Read at most maximum bytes plus one sentinel byte."""
process: subprocess.Popen[bytes] | None = None
try:
process = subprocess.Popen(
["git", "--git-dir", str(path), "cat-file", "blob", oid],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
env=RepositoryDiscovery._git_env(),
)
assert process.stdout is not None
content = process.stdout.read(maximum + 1)
truncated = len(content) > maximum
if truncated:
process.kill()
process.wait(timeout=1)
return content[:maximum], True
if process.wait(timeout=_GIT_TIMEOUT_SECONDS) != 0:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED")
return content, False
except RepositoryDiscoveryError:
raise
except (OSError, subprocess.TimeoutExpired) as exc:
if process is not None:
try:
process.kill()
process.wait(timeout=1)
except (OSError, subprocess.TimeoutExpired):
pass
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED") from exc
def _iter_nul_records(self, path: Path, args: list[str]) -> Iterator[bytes]:
if not args or args[0] not in _READ_ONLY_COMMANDS:
raise _error("REPOSITORY_DISCOVERY_GIT_COMMAND_REJECTED")
process: subprocess.Popen[bytes] | None = None
buffer = b""
started = time.monotonic()
try:
process = subprocess.Popen(
["git", "--git-dir", str(path), *args],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
env=self._git_env(),
)
assert process.stdout is not None
while True:
if time.monotonic() - started > _GIT_TIMEOUT_SECONDS:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED")
chunk = process.stdout.read(64 * 1024)
if not chunk:
break
buffer += chunk
if len(buffer) > _MAX_RECORD_BYTES and b"\0" not in buffer:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
while b"\0" in buffer:
record, buffer = buffer.split(b"\0", 1)
if len(record) > _MAX_RECORD_BYTES:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
yield record
if buffer:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
if process.wait(
timeout=max(0.1, _GIT_TIMEOUT_SECONDS - (time.monotonic() - started))
) != 0:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED")
except RepositoryDiscoveryError:
raise
except (OSError, subprocess.TimeoutExpired) as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED") from exc
finally:
if process is not None and process.poll() is None:
try:
process.kill()
process.wait(timeout=1)
except (OSError, subprocess.TimeoutExpired):
pass
def _resolve_at(self, path: Path, revision: object, field: str) -> str:
if (
not isinstance(revision, str)
or revision != revision.strip()
or not revision
or revision.startswith("-")
):
raise _error("REPOSITORY_DISCOVERY_INVALID_REVISION")
if _OID_RE.fullmatch(revision.lower()):
candidates = [revision.lower()]
elif revision == "HEAD":
candidates = ["HEAD"]
elif _safe_ref(revision) and not any(
token in revision for token in ("^", "~", ":", "\\")
):
if revision.startswith("refs/"):
if not revision.startswith(("refs/heads/", "refs/tags/")):
raise _error("REPOSITORY_DISCOVERY_INVALID_REVISION")
candidates = [revision]
else:
candidates = [f"refs/heads/{revision}", f"refs/tags/{revision}"]
else:
raise _error("REPOSITORY_DISCOVERY_INVALID_REVISION")
resolved: list[str] = []
for candidate in candidates:
try:
output = self._run(
path,
["rev-parse", "--verify", f"{candidate}^{{commit}}"],
max_output=256,
)
except RepositoryDiscoveryError:
continue
oid = output.decode("ascii", errors="ignore").strip().lower()
if _OID_RE.fullmatch(oid):
resolved.append(oid)
if not resolved:
raise RepositoryDiscoveryNotFound(
f"REPOSITORY_DISCOVERY_{field.upper()}_NOT_FOUND"
)
if len(set(resolved)) != 1:
raise _error("REPOSITORY_DISCOVERY_AMBIGUOUS_REVISION")
return resolved[0]
def resolve_revision(self, repository: str, revision: str) -> str:
return self._resolve_at(self._repository_path(repository), revision, "revision")
def _cursor_offset(
self,
cursor: str | None,
*,
operation: str,
repository: str,
resolved_oid: str,
query: Mapping[str, object],
) -> int:
if cursor is None:
return 0
if (
not isinstance(cursor, str)
or not cursor
or len(cursor) > _MAX_CURSOR_LENGTH
or "." not in cursor
):
raise _error("REPOSITORY_DISCOVERY_CURSOR_INVALID")
encoded, supplied_signature = cursor.split(".", 1)
try:
padded = encoded + "=" * (-len(encoded) % 4)
payload = json.loads(
base64.urlsafe_b64decode(padded.encode("ascii")).decode("ascii")
)
except (UnicodeError, ValueError, json.JSONDecodeError) as exc:
raise _error("REPOSITORY_DISCOVERY_CURSOR_INVALID") from exc
expected_signature = hmac.new(
self._cursor_secret, encoded.encode("ascii"), hashlib.sha256
).hexdigest()
if not hmac.compare_digest(supplied_signature, expected_signature):
raise _error("REPOSITORY_DISCOVERY_CURSOR_INVALID")
expected = {
"operation": operation,
"repository": repository,
"resolved_oid": resolved_oid,
"query": dict(query),
}
if (
not isinstance(payload, dict)
or payload.get("v") != 1
or any(payload.get(key) != value for key, value in expected.items())
):
raise _error("REPOSITORY_DISCOVERY_CURSOR_INVALID")
offset = payload.get("offset")
if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
raise _error("REPOSITORY_DISCOVERY_CURSOR_INVALID")
return offset
def _next_cursor(
self,
*,
operation: str,
repository: str,
resolved_oid: str,
query: Mapping[str, object],
offset: int,
) -> str:
payload = {
"v": 1,
"operation": operation,
"repository": repository,
"resolved_oid": resolved_oid,
"query": dict(query),
"offset": offset,
}
encoded = base64.urlsafe_b64encode(_canonical(payload)).decode("ascii").rstrip("=")
signature = hmac.new(
self._cursor_secret, encoded.encode("ascii"), hashlib.sha256
).hexdigest()
return f"{encoded}.{signature}"
def _iter_tree_entries(
self,
path: Path,
oid: str,
prefix: str,
*,
include_trees: bool,
) -> Iterator[dict[str, str]]:
args = ["ls-tree", "-z", "-r", "--full-tree"]
if include_trees:
args.append("-t")
args.append(oid)
if prefix:
args.extend(["--", prefix])
for record in self._iter_nul_records(path, args):
entry = _parse_tree_record(record)
if entry is None:
continue
if not prefix or entry["path"] == prefix or entry["path"].startswith(prefix + "/"):
yield entry
def list_tree(
self,
*,
repository: str,
revision: str,
prefix: str = "",
depth: int = 8,
limit: int = 100,
cursor: str | None = None,
) -> dict[str, object]:
prefix = _safe_relative_path(prefix, "prefix", allow_empty=True)
depth = _positive_int(depth, "depth", _MAX_DEPTH)
limit = _positive_int(limit, "limit", _MAX_LIMIT)
path = self._repository_path(repository)
resolved_oid = self._resolve_at(path, revision, "revision")
query = {"prefix": prefix, "depth": depth, "limit": limit}
offset = self._cursor_offset(
cursor,
operation="list_tree",
repository=repository,
resolved_oid=resolved_oid,
query=query,
)
page: list[dict[str, str]] = []
matched = 0
for entry in self._iter_tree_entries(path, resolved_oid, prefix, include_trees=True):
relative = entry["path"][len(prefix):].lstrip("/") if prefix else entry["path"]
if relative.count("/") + 1 > depth:
continue
if matched >= offset and len(page) < limit:
page.append(entry)
matched += 1
has_more = offset + len(page) < matched
return {
"repository": repository,
"resolved_oid": resolved_oid,
"entries": page,
"truncated": has_more,
"next_cursor": (
self._next_cursor(
operation="list_tree",
repository=repository,
resolved_oid=resolved_oid,
query=query,
offset=offset + len(page),
)
if has_more
else None
),
}
def list_repository_tree(self, request: Mapping[str, object]) -> dict[str, object]:
request = _request_mapping(request)
return self.list_tree(
repository=_required_request_value(request, "repository"), # type: ignore[arg-type]
revision=request.get("revision", request.get("ref", "HEAD")), # type: ignore[arg-type]
prefix=request.get("prefix", ""), # type: ignore[arg-type]
depth=request.get("depth", 8), # type: ignore[arg-type]
limit=request.get("limit", 100), # type: ignore[arg-type]
cursor=request.get("cursor"), # type: ignore[arg-type]
)
def _blob_size(self, path: Path, oid: str) -> int:
output = self._run(path, ["cat-file", "-s", oid], max_output=128)
try:
size = int(output.decode("ascii").strip())
except (UnicodeDecodeError, ValueError) as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
if size < 0:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
return size
def search_source(
self,
*,
repository: str,
revision: str,
query: str,
path_globs: Iterable[str] | None = None,
max_matches: int = 100,
max_bytes: int = 262_144,
cursor: str | None = None,
) -> dict[str, object]:
if (
not isinstance(query, str)
or not query
or "\x00" in query
or len(query) > _MAX_QUERY_LENGTH
):
raise _error("REPOSITORY_DISCOVERY_QUERY_INVALID")
max_matches = _positive_int(max_matches, "max_matches", _MAX_LIMIT)
max_bytes = _positive_int(max_bytes, "max_bytes", _MAX_BYTES)
if path_globs is None:
globs = ("*",)
else:
if isinstance(path_globs, (str, bytes)) or not isinstance(path_globs, Iterable):
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
globs = tuple(_safe_glob(item) for item in path_globs)
if not globs or len(globs) > _MAX_GLOBS:
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
path = self._repository_path(repository)
resolved_oid = self._resolve_at(path, revision, "revision")
query_binding = {
"query": query,
"path_globs": list(globs),
"max_matches": max_matches,
"max_bytes": max_bytes,
}
offset = self._cursor_offset(
cursor,
operation="search_source",
repository=repository,
resolved_oid=resolved_oid,
query=query_binding,
)
scanned_bytes = 0
skipped_for_bound = False
binary_blobs = 0
matched = 0
matches: list[dict[str, object]] = []
has_more = False
for entry in self._iter_tree_entries(path, resolved_oid, "", include_trees=False):
if entry["type"] != "blob" or not any(
fnmatch.fnmatchcase(entry["path"], pattern) for pattern in globs
):
continue
blob_size = self._blob_size(path, entry["object_oid"])
remaining = max_bytes - scanned_bytes
if blob_size > remaining:
skipped_for_bound = True
continue
raw, clipped = self._read_blob_prefix(path, entry["object_oid"], blob_size)
if clipped:
raise _error("REPOSITORY_DISCOVERY_GIT_FAILED")
scanned_bytes += len(raw)
if b"\0" in raw:
binary_blobs += 1
continue
text = raw.decode("utf-8", errors="replace")
for number, line in enumerate(text.splitlines(), start=1):
if query not in line:
continue
if matched < offset:
matched += 1
continue
if len(matches) >= max_matches:
has_more = True
break
matches.append(
{
"path": entry["path"],
"line_start": number,
"line_end": number,
"blob_oid": entry["object_oid"],
"snippet": line[:512],
}
)
matched += 1
if has_more:
break
return {
"repository": repository,
"resolved_oid": resolved_oid,
"matches": matches,
"scanned_bytes": scanned_bytes,
"max_bytes": max_bytes,
"binary_blobs_skipped": binary_blobs,
"truncated": bool(has_more or skipped_for_bound),
"next_cursor": (
self._next_cursor(
operation="search_source",
repository=repository,
resolved_oid=resolved_oid,
query=query_binding,
offset=offset + len(matches),
)
if has_more
else None
),
}
def search_repository_source(self, request: Mapping[str, object]) -> dict[str, object]:
request = _request_mapping(request)
path_globs = request.get("path_globs")
if path_globs is not None and (
isinstance(path_globs, (str, bytes)) or not isinstance(path_globs, Iterable)
):
raise _error("REPOSITORY_DISCOVERY_PATH_INVALID")
return self.search_source(
repository=_required_request_value(request, "repository"), # type: ignore[arg-type]
revision=request.get("revision", request.get("ref", "HEAD")), # type: ignore[arg-type]
query=_required_request_value(request, "query"), # type: ignore[arg-type]
path_globs=path_globs, # type: ignore[arg-type]
max_matches=request.get("max_matches", 100), # type: ignore[arg-type]
max_bytes=request.get("max_bytes", 262_144), # type: ignore[arg-type]
cursor=request.get("cursor"), # type: ignore[arg-type]
)
def _iter_diff_name_entries(
self,
path: Path,
from_oid: str,
to_oid: str,
prefix: str,
) -> Iterator[dict[str, object]]:
suffix = ["--", prefix] if prefix else []
records = self._iter_nul_records(
path,
[
"diff",
"--no-ext-diff",
"--no-textconv",
"--find-renames=50%",
"--name-status",
"-z",
from_oid,
to_oid,
*suffix,
],
)
iterator = iter(records)
while True:
try:
status_raw = next(iterator)
except StopIteration:
return
try:
status = status_raw.decode("ascii")
except UnicodeDecodeError as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
if not status or status[0] not in {"A", "C", "D", "M", "R", "T", "U", "X"}:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
try:
if status[0] in {"R", "C"}:
old_path = _decode_output_path(next(iterator))
new_path = _decode_output_path(next(iterator))
yield {
"status": status,
"path": new_path,
"old_path": old_path,
}
else:
yield {
"status": status,
"path": _decode_output_path(next(iterator)),
"old_path": None,
}
except StopIteration as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
@staticmethod
def _parse_numstat_count(value: bytes) -> int | None:
if value == b"-":
return None
try:
parsed = int(value.decode("ascii"), 10)
except (UnicodeDecodeError, ValueError) as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
if parsed < 0:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
return parsed
def _diff_stats(
self,
path: Path,
from_oid: str,
to_oid: str,
prefix: str,
wanted_paths: set[str],
) -> dict[str, tuple[int | None, int | None]]:
suffix = ["--", prefix] if prefix else []
records = self._iter_nul_records(
path,
[
"diff",
"--no-ext-diff",
"--no-textconv",
"--find-renames=50%",
"--numstat",
"-z",
from_oid,
to_oid,
*suffix,
],
)
iterator = iter(records)
stats: dict[str, tuple[int | None, int | None]] = {}
for row in iterator:
if not row:
continue
fields = row.split(b"\t", 2)
if len(fields) != 3:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID")
additions = self._parse_numstat_count(fields[0])
deletions = self._parse_numstat_count(fields[1])
if fields[2]:
stat_path = _decode_output_path(fields[2])
else:
try:
next(iterator) # old path
stat_path = _decode_output_path(next(iterator))
except StopIteration as exc:
raise _error("REPOSITORY_DISCOVERY_GIT_OUTPUT_INVALID") from exc
if stat_path in wanted_paths:
stats[stat_path] = (additions, deletions)
if wanted_paths <= set(stats):
break
return stats
def diff_index(
self,
*,
repository: str,
from_revision: str,
to_revision: str,
prefix: str = "",
limit: int = 100,
cursor: str | None = None,
) -> dict[str, object]:
prefix = _safe_relative_path(prefix, "prefix", allow_empty=True)
limit = _positive_int(limit, "limit", _MAX_LIMIT)
path = self._repository_path(repository)
from_oid = self._resolve_at(path, from_revision, "from")
to_oid = self._resolve_at(path, to_revision, "to")
query = {
"from_oid": from_oid,
"to_oid": to_oid,
"prefix": prefix,
"limit": limit,
}
offset = self._cursor_offset(
cursor,
operation="diff_index",
repository=repository,
resolved_oid=to_oid,
query=query,
)
page: list[dict[str, object]] = []
total = 0
for entry in self._iter_diff_name_entries(path, from_oid, to_oid, prefix):
if total >= offset and len(page) < limit:
page.append(entry)
total += 1
wanted_paths = {str(entry["path"]) for entry in page}
stat_by_path = (
self._diff_stats(path, from_oid, to_oid, prefix, wanted_paths)
if page
else {}
)
for entry in page:
additions, deletions = stat_by_path.get(
str(entry["path"]), (None, None)
)
entry["additions"] = additions
entry["deletions"] = deletions
has_more = offset + len(page) < total
return {
"repository": repository,
"resolved_from_oid": from_oid,
"resolved_to_oid": to_oid,
"entries": page,
"truncated": has_more,
"next_cursor": (
self._next_cursor(
operation="diff_index",
repository=repository,
resolved_oid=to_oid,
query=query,
offset=offset + len(page),
)
if has_more
else None
),
}
def get_diff_index(self, request: Mapping[str, object]) -> dict[str, object]:
request = _request_mapping(request)
from_revision = request.get("from_revision", request.get("from"))
to_revision = request.get("to_revision", request.get("to"))
prefix = request.get("prefix", request.get("path_prefix", ""))
if from_revision is None or to_revision is None:
raise _error("REPOSITORY_DISCOVERY_REQUEST_INVALID")
return self.diff_index(
repository=_required_request_value(request, "repository"), # type: ignore[arg-type]
from_revision=from_revision, # type: ignore[arg-type]
to_revision=to_revision, # type: ignore[arg-type]
prefix=prefix, # type: ignore[arg-type]
limit=request.get("limit", 100), # type: ignore[arg-type]
cursor=request.get("cursor"), # type: ignore[arg-type]
)
def read_file(
self,
*,
repository: str,
revision: str,
file_path: str,
max_bytes: int = 262_144,
) -> dict[str, object]:
file_path = _safe_relative_path(file_path, "file_path", allow_empty=False)
max_bytes = _positive_int(max_bytes, "max_bytes", _MAX_BYTES)
path = self._repository_path(repository)
resolved_oid = self._resolve_at(path, revision, "revision")
exact: list[dict[str, str]] = []
for entry in self._iter_tree_entries(
path, resolved_oid, file_path, include_trees=False
):
if entry["path"] == file_path and entry["type"] == "blob":
exact.append(entry)
if len(exact) != 1:
raise RepositoryDiscoveryNotFound("REPOSITORY_DISCOVERY_FILE_NOT_FOUND")
blob_oid = exact[0]["object_oid"]
content, truncated = self._read_blob_prefix(path, blob_oid, max_bytes)
return {
"repository": repository,
"resolved_oid": resolved_oid,
"blob_oid": blob_oid,
"path": file_path,
"content": content.decode("utf-8", errors="replace"),
"content_base64": base64.b64encode(content).decode("ascii"),
"content_bytes": len(content),
"truncated": truncated,
}
def read_repository_file(self, request: Mapping[str, object]) -> dict[str, object]:
request = _request_mapping(request)
file_path = request.get("file_path", request.get("path"))
if file_path is None:
raise _error("REPOSITORY_DISCOVERY_REQUEST_INVALID")
return self.read_file(
repository=_required_request_value(request, "repository"), # type: ignore[arg-type]
revision=request.get("revision", request.get("ref", "HEAD")), # type: ignore[arg-type]
file_path=file_path, # type: ignore[arg-type]
max_bytes=request.get("max_bytes", 262_144), # type: ignore[arg-type]
)
def create_repository_discovery(
*,
allowed_repositories: Iterable[str] | None = None,
git_dir: str | Path | None = None,
cursor_secret: str | bytes | None = None,
) -> RepositoryDiscovery:
return RepositoryDiscovery(
allowed_repositories=allowed_repositories,
git_dir=git_dir,
cursor_secret=cursor_secret,
)
def list_repository_tree(
request: Mapping[str, object], *, discovery: RepositoryDiscovery | None = None
) -> dict[str, object]:
return (discovery or create_repository_discovery()).list_repository_tree(request)
def search_repository_source(
request: Mapping[str, object], *, discovery: RepositoryDiscovery | None = None
) -> dict[str, object]:
return (discovery or create_repository_discovery()).search_repository_source(request)
def get_diff_index(
request: Mapping[str, object], *, discovery: RepositoryDiscovery | None = None
) -> dict[str, object]:
return (discovery or create_repository_discovery()).get_diff_index(request)
def read_repository_file(
request: Mapping[str, object], *, discovery: RepositoryDiscovery | None = None
) -> dict[str, object]:
return (discovery or create_repository_discovery()).read_repository_file(request)
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,707 @@
"""Single-host operational delegation for AIOS migration.
This module is deliberately narrow:
* Owner signs only OperationalDelegation and ExpiredTaskRecoveryDecision.
* Machine signer signs only WorkerRuntimeBinding and WorkerRuntimeProof.
* Every object is bound to one immutable Task identity and a narrow capability set.
* There is no arbitrary sign-bytes API.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
import hashlib
import json
from pathlib import Path
import re
import time
from typing import Any, Mapping, Sequence
from aios_single_host_authority import (
DEFAULT_MACHINE_SIGNER_ID,
DEFAULT_OWNER_SIGNER_ID,
MACHINE_NAMESPACE,
OWNER_IDENTITY,
OWNER_NAMESPACE,
LocalFileSigner,
RegistryOwnerApprovalVerifier,
SignerRegistry,
SingleHostAuthorityError,
canonical_json_bytes,
sha256_json,
verify_sshsig,
)
OPERATIONAL_POLICY_SCHEMA = "aios.single-host-operational-policy.v1"
OPERATIONAL_DELEGATION_SCHEMA = "aios.owner-operational-delegation.v1"
RECOVERY_DECISION_SCHEMA = "aios.expired-task-recovery-decision.v1"
WORKER_BINDING_SCHEMA = "aios.worker-runtime-binding.v1"
WORKER_PROOF_SCHEMA = "aios.worker-runtime-proof.v1"
REVOCATION_SCHEMA = "aios.operational-revocations.v1"
ALLOWED_OPERATIONAL_CAPABILITIES = frozenset({
"task.recovery",
"worker.runtime.bind",
"worker.runtime.proof",
"task.claim.guard",
})
ALLOWED_EXPECTED_STATE = "EXPIRED"
HEX40 = re.compile(r"^[0-9a-f]{40}$")
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/+:-]{2,511}$")
CAP = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
class OperationalAuthorityError(RuntimeError):
def __init__(self, code: str):
super().__init__(code)
self.code = code
def fail(code: str) -> None:
raise OperationalAuthorityError(code)
def _exact_fields(value: Mapping[str, Any], fields: set[str], code: str) -> None:
if not isinstance(value, Mapping) or set(value) != fields:
fail(code)
def _safe_id(value: str, code: str) -> str:
if not isinstance(value, str) or not SAFE_ID.fullmatch(value):
fail(code)
return value
def _hex(value: str, n: int, code: str) -> str:
regex = HEX40 if n == 40 else HEX64
if not isinstance(value, str) or not regex.fullmatch(value):
fail(code)
return value
def _caps(value: Sequence[str], code: str) -> tuple[str, ...]:
if not isinstance(value, (list, tuple)) or not value:
fail(code)
out = tuple(sorted(set(value)))
if len(out) != len(value) or any(not isinstance(x, str) or not CAP.fullmatch(x) for x in out):
fail(code)
return out
def task_identity_sha256(task: Mapping[str, Any]) -> str:
"""Digest only immutable recovery/claim identity fields.
Revision, state, attempt and deadline are intentionally excluded.
"""
fields = (
"task_id", "repository", "target_branch", "base_oid", "generation",
"execution_plan_sha256", "external_job_identity",
)
try:
value = {name: task[name] for name in fields}
except KeyError as exc:
raise OperationalAuthorityError("TASK_IDENTITY_FIELD_MISSING") from exc
_safe_id(str(value["task_id"]), "TASK_ID_INVALID")
_safe_id(str(value["repository"]), "REPOSITORY_INVALID")
_safe_id(str(value["target_branch"]), "TARGET_BRANCH_INVALID")
_hex(str(value["base_oid"]), 40, "BASE_OID_INVALID")
if type(value["generation"]) is not int or value["generation"] < 1:
fail("GENERATION_INVALID")
_hex(str(value["execution_plan_sha256"]), 64, "EXECUTION_PLAN_DIGEST_INVALID")
_safe_id(str(value["external_job_identity"]), "EXTERNAL_JOB_IDENTITY_INVALID")
return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
@dataclass(frozen=True, slots=True)
class OperationalPolicy:
repository: str
task_id: str
task_identity_sha256: str
target_branch: str
base_oid: str
generation: int
execution_plan_sha256: str
external_job_identity: str
allowed_capabilities: tuple[str, ...]
allowed_worker_ids: tuple[str, ...]
maximum_delegation_ttl_seconds: int
maximum_recovery_decision_ttl_seconds: int
maximum_binding_ttl_seconds: int
maximum_proof_ttl_seconds: int
maximum_heartbeat_age_seconds: int
maximum_task_deadline_extension_seconds: int
schema: str = OPERATIONAL_POLICY_SCHEMA
def validate(self) -> None:
if self.schema != OPERATIONAL_POLICY_SCHEMA:
fail("OPERATIONAL_POLICY_SCHEMA_INVALID")
_safe_id(self.repository, "OPERATIONAL_POLICY_REPOSITORY_INVALID")
_safe_id(self.task_id, "OPERATIONAL_POLICY_TASK_ID_INVALID")
_hex(self.task_identity_sha256, 64, "OPERATIONAL_POLICY_TASK_IDENTITY_INVALID")
_safe_id(self.target_branch, "OPERATIONAL_POLICY_BRANCH_INVALID")
_hex(self.base_oid, 40, "OPERATIONAL_POLICY_BASE_OID_INVALID")
if type(self.generation) is not int or self.generation < 1:
fail("OPERATIONAL_POLICY_GENERATION_INVALID")
_hex(self.execution_plan_sha256, 64, "OPERATIONAL_POLICY_PLAN_INVALID")
_safe_id(self.external_job_identity, "OPERATIONAL_POLICY_JOB_INVALID")
caps = _caps(self.allowed_capabilities, "OPERATIONAL_POLICY_CAPABILITIES_INVALID")
if not set(caps).issubset(ALLOWED_OPERATIONAL_CAPABILITIES):
fail("OPERATIONAL_POLICY_CAPABILITY_WIDENING")
if not self.allowed_worker_ids or any(not SAFE_ID.fullmatch(x) for x in self.allowed_worker_ids):
fail("OPERATIONAL_POLICY_WORKERS_INVALID")
bounds = (
(self.maximum_delegation_ttl_seconds, 300, 7 * 24 * 3600),
(self.maximum_recovery_decision_ttl_seconds, 30, 900),
(self.maximum_binding_ttl_seconds, 60, 24 * 3600),
(self.maximum_proof_ttl_seconds, 15, 900),
(self.maximum_heartbeat_age_seconds, 5, 300),
(self.maximum_task_deadline_extension_seconds, 300, 24 * 3600),
)
if any(type(v) is not int or not lo <= v <= hi for v, lo, hi in bounds):
fail("OPERATIONAL_POLICY_TTL_INVALID")
@property
def digest(self) -> str:
self.validate()
return sha256_json(asdict(self))
@classmethod
def from_mapping(cls, value: Mapping[str, Any]) -> "OperationalPolicy":
fields = set(cls.__dataclass_fields__)
_exact_fields(value, fields, "OPERATIONAL_POLICY_FIELDS_INVALID")
try:
policy = cls(
repository=value["repository"],
task_id=value["task_id"],
task_identity_sha256=value["task_identity_sha256"],
target_branch=value["target_branch"],
base_oid=value["base_oid"],
generation=value["generation"],
execution_plan_sha256=value["execution_plan_sha256"],
external_job_identity=value["external_job_identity"],
allowed_capabilities=tuple(value["allowed_capabilities"]),
allowed_worker_ids=tuple(value["allowed_worker_ids"]),
maximum_delegation_ttl_seconds=value["maximum_delegation_ttl_seconds"],
maximum_recovery_decision_ttl_seconds=value["maximum_recovery_decision_ttl_seconds"],
maximum_binding_ttl_seconds=value["maximum_binding_ttl_seconds"],
maximum_proof_ttl_seconds=value["maximum_proof_ttl_seconds"],
maximum_heartbeat_age_seconds=value["maximum_heartbeat_age_seconds"],
maximum_task_deadline_extension_seconds=value["maximum_task_deadline_extension_seconds"],
schema=value["schema"],
)
except Exception as exc:
raise OperationalAuthorityError("OPERATIONAL_POLICY_FIELDS_INVALID") from exc
policy.validate()
return policy
def build_exact_policy(
task: Mapping[str, Any],
*,
worker_id: str,
allowed_capabilities: Sequence[str] = tuple(sorted(ALLOWED_OPERATIONAL_CAPABILITIES)),
) -> OperationalPolicy:
identity = task_identity_sha256(task)
policy = OperationalPolicy(
repository=str(task["repository"]),
task_id=str(task["task_id"]),
task_identity_sha256=identity,
target_branch=str(task["target_branch"]),
base_oid=str(task["base_oid"]),
generation=int(task["generation"]),
execution_plan_sha256=str(task["execution_plan_sha256"]),
external_job_identity=str(task["external_job_identity"]),
allowed_capabilities=tuple(sorted(allowed_capabilities)),
allowed_worker_ids=(worker_id,),
maximum_delegation_ttl_seconds=24 * 3600,
maximum_recovery_decision_ttl_seconds=300,
maximum_binding_ttl_seconds=4 * 3600,
maximum_proof_ttl_seconds=300,
maximum_heartbeat_age_seconds=30,
maximum_task_deadline_extension_seconds=4 * 3600,
)
policy.validate()
return policy
def _signed_payload_dict(payload: Mapping[str, Any], *, key_id: str, namespace: str) -> dict[str, Any]:
value = dict(payload)
value["signature_namespace"] = namespace
value["signature_key_id"] = key_id
value["canonical_payload_digest"] = sha256_json(value)
return value
def _verify_digest_field(payload: Mapping[str, Any]) -> bool:
if not isinstance(payload, Mapping) or "canonical_payload_digest" not in payload:
return False
value = dict(payload)
supplied = value.pop("canonical_payload_digest", None)
return isinstance(supplied, str) and supplied == sha256_json(value)
@dataclass(frozen=True, slots=True)
class SignedOperationalArtifact:
payload: dict[str, Any]
signature: str
key_id: str
namespace: str
@property
def canonical_payload(self) -> bytes:
return canonical_json_bytes(self.payload)
@property
def payload_sha256(self) -> str:
return hashlib.sha256(self.canonical_payload).hexdigest()
def to_dict(self) -> dict[str, Any]:
return {
"payload": dict(self.payload),
"signature": self.signature,
"key_id": self.key_id,
"namespace": self.namespace,
}
class OperationalOwnerAuthority:
def __init__(self, key_path: Path, policy: OperationalPolicy, *, require_root_owner: bool = True, clock: Any = time.time):
policy.validate()
self.policy = policy
self.signer = LocalFileSigner(
Path(key_path),
signer_id=DEFAULT_OWNER_SIGNER_ID,
namespace=OWNER_NAMESPACE,
require_root_owner=require_root_owner,
)
self.clock = clock
@property
def fingerprint(self) -> str:
return self.signer.fingerprint
def _sign(self, payload: Mapping[str, Any]) -> SignedOperationalArtifact:
full = _signed_payload_dict(payload, key_id=self.fingerprint, namespace=OWNER_NAMESPACE)
art = self.signer._sign_canonical(full)
return SignedOperationalArtifact(dict(art.payload), art.signature, art.key_id, art.namespace)
def issue_operational_delegation(self, *, delegation_id: str, nonce: str, ttl_seconds: int) -> SignedOperationalArtifact:
now = int(self.clock())
p = self.policy
if not SAFE_ID.fullmatch(delegation_id) or not SAFE_ID.fullmatch(nonce):
fail("OPERATIONAL_DELEGATION_ID_INVALID")
if type(ttl_seconds) is not int or not 60 <= ttl_seconds <= p.maximum_delegation_ttl_seconds:
fail("OPERATIONAL_DELEGATION_TTL_INVALID")
payload = {
"schema": OPERATIONAL_DELEGATION_SCHEMA,
"delegation_id": delegation_id,
"nonce": nonce,
"owner_identity": OWNER_IDENTITY,
"repository": p.repository,
"task_id": p.task_id,
"task_identity_sha256": p.task_identity_sha256,
"target_branch": p.target_branch,
"base_oid": p.base_oid,
"generation": p.generation,
"execution_plan_sha256": p.execution_plan_sha256,
"external_job_identity": p.external_job_identity,
"allowed_capabilities": list(p.allowed_capabilities),
"allowed_worker_ids": list(p.allowed_worker_ids),
"policy_digest": p.digest,
"issued_at": now,
"not_before": max(0, now - 5),
"expires_at": now + ttl_seconds,
"revocation_epoch": 0,
"status": "ACTIVE",
}
return self._sign(payload)
def issue_recovery_decision(
self,
delegation: Mapping[str, Any],
*,
decision_id: str,
nonce: str,
expected_revision: int,
expected_attempt: int,
original_deadline: int,
new_deadline: int,
ttl_seconds: int,
) -> SignedOperationalArtifact:
now = int(self.clock())
verify_delegation_scope(delegation, self.policy, now=now)
if "task.recovery" not in delegation["allowed_capabilities"]:
fail("RECOVERY_CAPABILITY_NOT_DELEGATED")
if not SAFE_ID.fullmatch(decision_id) or not SAFE_ID.fullmatch(nonce):
fail("RECOVERY_DECISION_ID_INVALID")
if type(expected_revision) is not int or expected_revision < 0 or type(expected_attempt) is not int or expected_attempt < 0:
fail("RECOVERY_CAS_INVALID")
if type(original_deadline) is not int or type(new_deadline) is not int:
fail("RECOVERY_DEADLINE_INVALID")
if new_deadline <= now or new_deadline <= original_deadline:
fail("RECOVERY_DEADLINE_INVALID")
if new_deadline - now > self.policy.maximum_task_deadline_extension_seconds:
fail("RECOVERY_DEADLINE_WIDENING")
if type(ttl_seconds) is not int or not 15 <= ttl_seconds <= self.policy.maximum_recovery_decision_ttl_seconds:
fail("RECOVERY_DECISION_TTL_INVALID")
payload = {
"schema": RECOVERY_DECISION_SCHEMA,
"decision_id": decision_id,
"nonce": nonce,
"parent_delegation_id": delegation["delegation_id"],
"parent_delegation_digest": hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest(),
"task_id": self.policy.task_id,
"task_identity_sha256": self.policy.task_identity_sha256,
"expected_state": ALLOWED_EXPECTED_STATE,
"expected_revision": expected_revision,
"expected_attempt": expected_attempt,
"original_deadline": original_deadline,
"new_deadline": new_deadline,
"repository": self.policy.repository,
"target_branch": self.policy.target_branch,
"base_oid": self.policy.base_oid,
"generation": self.policy.generation,
"execution_plan_sha256": self.policy.execution_plan_sha256,
"external_job_identity": self.policy.external_job_identity,
"policy_digest": self.policy.digest,
"one_time_use": True,
"issued_at": now,
"not_before": max(0, now - 5),
"expires_at": now + ttl_seconds,
}
return self._sign(payload)
class OperationalMachineAuthority:
def __init__(self, key_path: Path, policy: OperationalPolicy, *, require_root_owner: bool = True, clock: Any = time.time):
policy.validate()
self.policy = policy
self.signer = LocalFileSigner(
Path(key_path),
signer_id=DEFAULT_MACHINE_SIGNER_ID,
namespace=MACHINE_NAMESPACE,
require_root_owner=require_root_owner,
)
self.clock = clock
@property
def fingerprint(self) -> str:
return self.signer.fingerprint
def _sign(self, payload: Mapping[str, Any]) -> SignedOperationalArtifact:
full = _signed_payload_dict(payload, key_id=self.fingerprint, namespace=MACHINE_NAMESPACE)
art = self.signer._sign_canonical(full)
return SignedOperationalArtifact(dict(art.payload), art.signature, art.key_id, art.namespace)
def issue_runtime_binding(
self,
delegation: Mapping[str, Any],
*,
binding_id: str,
nonce: str,
worker_id: str,
runtime_id: str,
runtime_sha256: str,
task_contract_version: str,
supported_contract_versions: Sequence[str],
required_capabilities: Sequence[str],
worker_capabilities: Sequence[str],
max_concurrency: int,
ttl_seconds: int,
) -> SignedOperationalArtifact:
now = int(self.clock())
p = self.policy
verify_delegation_scope(delegation, p, now=now)
if "worker.runtime.bind" not in delegation["allowed_capabilities"]:
fail("RUNTIME_BIND_CAPABILITY_NOT_DELEGATED")
if worker_id not in p.allowed_worker_ids:
fail("WORKER_ID_NOT_DELEGATED")
for val, code in [(binding_id, "BINDING_ID_INVALID"), (nonce, "BINDING_NONCE_INVALID"), (worker_id, "WORKER_ID_INVALID"), (runtime_id, "RUNTIME_ID_INVALID"), (task_contract_version, "CONTRACT_VERSION_INVALID")]:
_safe_id(val, code)
_hex(runtime_sha256, 64, "RUNTIME_SHA256_INVALID")
supported = tuple(sorted(set(supported_contract_versions)))
if task_contract_version not in supported or not supported or any(not SAFE_ID.fullmatch(x) for x in supported):
fail("CONTRACT_INCOMPATIBLE")
required = _caps(required_capabilities, "REQUIRED_CAPABILITIES_INVALID")
worker_caps = _caps(worker_capabilities, "WORKER_CAPABILITIES_INVALID")
if not set(required).issubset(set(worker_caps)):
fail("CAPABILITY_MISMATCH")
if type(max_concurrency) is not int or not 1 <= max_concurrency <= 64:
fail("MAX_CONCURRENCY_INVALID")
if type(ttl_seconds) is not int or not 30 <= ttl_seconds <= p.maximum_binding_ttl_seconds:
fail("BINDING_TTL_INVALID")
if now + ttl_seconds > int(delegation["expires_at"]):
fail("BINDING_EXCEEDS_PARENT")
payload = {
"schema": WORKER_BINDING_SCHEMA,
"binding_id": binding_id,
"nonce": nonce,
"parent_delegation_id": delegation["delegation_id"],
"parent_delegation_digest": hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest(),
"repository": p.repository,
"task_id": p.task_id,
"task_identity_sha256": p.task_identity_sha256,
"target_branch": p.target_branch,
"base_oid": p.base_oid,
"generation": p.generation,
"execution_plan_sha256": p.execution_plan_sha256,
"external_job_identity": p.external_job_identity,
"worker_id": worker_id,
"runtime_id": runtime_id,
"runtime_sha256": runtime_sha256,
"task_contract_version": task_contract_version,
"supported_contract_versions": list(supported),
"required_capabilities": list(required),
"worker_capabilities": list(worker_caps),
"max_concurrency": max_concurrency,
"policy_digest": p.digest,
"issued_at": now,
"not_before": max(0, now - 5),
"expires_at": now + ttl_seconds,
}
return self._sign(payload)
def issue_runtime_proof(
self,
delegation: Mapping[str, Any],
binding: Mapping[str, Any],
*,
proof_id: str,
nonce: str,
heartbeat_at: int,
adapter_readiness: bool,
resource_readiness: bool,
active_claim_count: int,
ttl_seconds: int,
) -> SignedOperationalArtifact:
now = int(self.clock())
p = self.policy
verify_delegation_scope(delegation, p, now=now)
verify_binding_scope(binding, delegation, p, now=now)
if "worker.runtime.proof" not in delegation["allowed_capabilities"]:
fail("RUNTIME_PROOF_CAPABILITY_NOT_DELEGATED")
_safe_id(proof_id, "PROOF_ID_INVALID")
_safe_id(nonce, "PROOF_NONCE_INVALID")
if type(heartbeat_at) is not int or heartbeat_at > now + 5 or now - heartbeat_at > p.maximum_heartbeat_age_seconds:
fail("HEARTBEAT_STALE")
if adapter_readiness is not True or resource_readiness is not True:
fail("RUNTIME_NOT_READY")
if type(active_claim_count) is not int or active_claim_count < 0 or active_claim_count >= int(binding["max_concurrency"]):
fail("MAX_CONCURRENCY_REACHED")
if type(ttl_seconds) is not int or not 10 <= ttl_seconds <= p.maximum_proof_ttl_seconds:
fail("PROOF_TTL_INVALID")
if now + ttl_seconds > min(int(binding["expires_at"]), int(delegation["expires_at"])):
fail("PROOF_EXCEEDS_PARENT")
payload = {
"schema": WORKER_PROOF_SCHEMA,
"proof_id": proof_id,
"nonce": nonce,
"parent_delegation_id": delegation["delegation_id"],
"parent_delegation_digest": hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest(),
"binding_id": binding["binding_id"],
"binding_digest": hashlib.sha256(canonical_json_bytes(dict(binding))).hexdigest(),
"repository": p.repository,
"task_id": p.task_id,
"task_identity_sha256": p.task_identity_sha256,
"generation": p.generation,
"execution_plan_sha256": p.execution_plan_sha256,
"worker_id": binding["worker_id"],
"runtime_id": binding["runtime_id"],
"runtime_sha256": binding["runtime_sha256"],
"heartbeat_at": heartbeat_at,
"adapter_readiness": True,
"resource_readiness": True,
"active_claim_count": active_claim_count,
"max_concurrency": binding["max_concurrency"],
"policy_digest": p.digest,
"issued_at": now,
"expires_at": now + ttl_seconds,
}
return self._sign(payload)
DELEGATION_FIELDS = {
"schema","delegation_id","nonce","owner_identity","repository","task_id",
"task_identity_sha256","target_branch","base_oid","generation","execution_plan_sha256",
"external_job_identity","allowed_capabilities","allowed_worker_ids","policy_digest",
"issued_at","not_before","expires_at","revocation_epoch","status",
"signature_namespace","signature_key_id","canonical_payload_digest",
}
BINDING_FIELDS = {
"schema","binding_id","nonce","parent_delegation_id","parent_delegation_digest",
"repository","task_id","task_identity_sha256","target_branch","base_oid","generation",
"execution_plan_sha256","external_job_identity","worker_id","runtime_id","runtime_sha256",
"task_contract_version","supported_contract_versions","required_capabilities","worker_capabilities",
"max_concurrency","policy_digest","issued_at","not_before","expires_at",
"signature_namespace","signature_key_id","canonical_payload_digest",
}
PROOF_FIELDS = {
"schema","proof_id","nonce","parent_delegation_id","parent_delegation_digest",
"binding_id","binding_digest","repository","task_id","task_identity_sha256","generation",
"execution_plan_sha256","worker_id","runtime_id","runtime_sha256","heartbeat_at",
"adapter_readiness","resource_readiness","active_claim_count","max_concurrency",
"policy_digest","issued_at","expires_at","signature_namespace","signature_key_id",
"canonical_payload_digest",
}
RECOVERY_FIELDS = {
"schema","decision_id","nonce","parent_delegation_id","parent_delegation_digest",
"task_id","task_identity_sha256","expected_state","expected_revision","expected_attempt",
"original_deadline","new_deadline","repository","target_branch","base_oid","generation",
"execution_plan_sha256","external_job_identity","policy_digest","one_time_use",
"issued_at","not_before","expires_at","signature_namespace","signature_key_id",
"canonical_payload_digest",
}
def verify_delegation_scope(value: Mapping[str, Any], policy: OperationalPolicy, *, now: int | None = None) -> None:
policy.validate()
_exact_fields(value, DELEGATION_FIELDS, "OPERATIONAL_DELEGATION_FIELDS_INVALID")
now = int(time.time()) if now is None else int(now)
if (
value["schema"] != OPERATIONAL_DELEGATION_SCHEMA
or value["status"] != "ACTIVE"
or value["revocation_epoch"] != 0
or value["owner_identity"] != OWNER_IDENTITY
or value["repository"] != policy.repository
or value["task_id"] != policy.task_id
or value["task_identity_sha256"] != policy.task_identity_sha256
or value["target_branch"] != policy.target_branch
or value["base_oid"] != policy.base_oid
or value["generation"] != policy.generation
or value["execution_plan_sha256"] != policy.execution_plan_sha256
or value["external_job_identity"] != policy.external_job_identity
or value["policy_digest"] != policy.digest
or not set(value["allowed_capabilities"]).issubset(set(policy.allowed_capabilities))
or not set(value["allowed_worker_ids"]).issubset(set(policy.allowed_worker_ids))
or value["signature_namespace"] != OWNER_NAMESPACE
or not _verify_digest_field(value)
or not int(value["not_before"]) <= now < int(value["expires_at"])
):
fail("OPERATIONAL_DELEGATION_SCOPE_INVALID")
def verify_binding_scope(binding: Mapping[str, Any], delegation: Mapping[str, Any], policy: OperationalPolicy, *, now: int | None = None) -> None:
_exact_fields(binding, BINDING_FIELDS, "WORKER_BINDING_FIELDS_INVALID")
now = int(time.time()) if now is None else int(now)
expected_parent = hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest()
if (
binding["schema"] != WORKER_BINDING_SCHEMA
or binding["parent_delegation_id"] != delegation["delegation_id"]
or binding["parent_delegation_digest"] != expected_parent
or binding["repository"] != policy.repository
or binding["task_id"] != policy.task_id
or binding["task_identity_sha256"] != policy.task_identity_sha256
or binding["target_branch"] != policy.target_branch
or binding["base_oid"] != policy.base_oid
or binding["generation"] != policy.generation
or binding["execution_plan_sha256"] != policy.execution_plan_sha256
or binding["external_job_identity"] != policy.external_job_identity
or binding["worker_id"] not in policy.allowed_worker_ids
or binding["policy_digest"] != policy.digest
or binding["signature_namespace"] != MACHINE_NAMESPACE
or not _verify_digest_field(binding)
or not int(binding["not_before"]) <= now < int(binding["expires_at"])
):
fail("WORKER_BINDING_SCOPE_INVALID")
required = set(binding["required_capabilities"])
worker = set(binding["worker_capabilities"])
if not required.issubset(worker) or binding["task_contract_version"] not in binding["supported_contract_versions"]:
fail("WORKER_BINDING_COMPATIBILITY_INVALID")
def verify_proof_scope(proof: Mapping[str, Any], binding: Mapping[str, Any], delegation: Mapping[str, Any], policy: OperationalPolicy, *, now: int | None = None) -> None:
_exact_fields(proof, PROOF_FIELDS, "WORKER_PROOF_FIELDS_INVALID")
now = int(time.time()) if now is None else int(now)
if (
proof["schema"] != WORKER_PROOF_SCHEMA
or proof["parent_delegation_id"] != delegation["delegation_id"]
or proof["parent_delegation_digest"] != hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest()
or proof["binding_id"] != binding["binding_id"]
or proof["binding_digest"] != hashlib.sha256(canonical_json_bytes(dict(binding))).hexdigest()
or proof["repository"] != policy.repository
or proof["task_id"] != policy.task_id
or proof["task_identity_sha256"] != policy.task_identity_sha256
or proof["generation"] != policy.generation
or proof["execution_plan_sha256"] != policy.execution_plan_sha256
or proof["worker_id"] != binding["worker_id"]
or proof["runtime_id"] != binding["runtime_id"]
or proof["runtime_sha256"] != binding["runtime_sha256"]
or proof["max_concurrency"] != binding["max_concurrency"]
or proof["policy_digest"] != policy.digest
or proof["signature_namespace"] != MACHINE_NAMESPACE
or proof["adapter_readiness"] is not True
or proof["resource_readiness"] is not True
or not _verify_digest_field(proof)
or not now < int(proof["expires_at"])
or now - int(proof["heartbeat_at"]) > policy.maximum_heartbeat_age_seconds
or int(proof["active_claim_count"]) >= int(proof["max_concurrency"])
):
fail("WORKER_PROOF_SCOPE_INVALID")
def verify_recovery_scope(decision: Mapping[str, Any], delegation: Mapping[str, Any], policy: OperationalPolicy, *, now: int | None = None) -> None:
_exact_fields(decision, RECOVERY_FIELDS, "RECOVERY_DECISION_FIELDS_INVALID")
now = int(time.time()) if now is None else int(now)
if (
decision["schema"] != RECOVERY_DECISION_SCHEMA
or decision["parent_delegation_id"] != delegation["delegation_id"]
or decision["parent_delegation_digest"] != hashlib.sha256(canonical_json_bytes(dict(delegation))).hexdigest()
or decision["task_id"] != policy.task_id
or decision["task_identity_sha256"] != policy.task_identity_sha256
or decision["expected_state"] != "EXPIRED"
or decision["repository"] != policy.repository
or decision["target_branch"] != policy.target_branch
or decision["base_oid"] != policy.base_oid
or decision["generation"] != policy.generation
or decision["execution_plan_sha256"] != policy.execution_plan_sha256
or decision["external_job_identity"] != policy.external_job_identity
or decision["policy_digest"] != policy.digest
or decision["one_time_use"] is not True
or decision["signature_namespace"] != OWNER_NAMESPACE
or not _verify_digest_field(decision)
or not int(decision["not_before"]) <= now < int(decision["expires_at"])
or int(decision["new_deadline"]) <= now
):
fail("RECOVERY_DECISION_SCOPE_INVALID")
class OperationalVerifier:
"""Verify signatures through the public root-owned SignerRegistry."""
def __init__(self, registry_path: Path, *, require_root_owner: bool = True):
self.registry_path = Path(registry_path)
self.require_root_owner = require_root_owner
def _verify(self, artifact: SignedOperationalArtifact | Mapping[str, Any], *, role: str, signer_id: str, namespace: str) -> dict[str, Any]:
if isinstance(artifact, SignedOperationalArtifact):
payload, signature, key_id, artifact_namespace = artifact.payload, artifact.signature, artifact.key_id, artifact.namespace
else:
_exact_fields(artifact, {"payload","signature","key_id","namespace"}, "SIGNED_ARTIFACT_FIELDS_INVALID")
payload, signature, key_id, artifact_namespace = artifact["payload"], artifact["signature"], artifact["key_id"], artifact["namespace"]
if artifact_namespace != namespace:
fail("SIGNED_ARTIFACT_NAMESPACE_INVALID")
registry = SignerRegistry(self.registry_path, require_root_owner=self.require_root_owner)
rec = registry.require(signer_id, role=role)
if rec.fingerprint != key_id:
fail("SIGNED_ARTIFACT_KEY_ID_INVALID")
if role == "MACHINE_OPERATIONAL":
owner_rec = registry.require(DEFAULT_OWNER_SIGNER_ID, role="OWNER_AUTHORITY")
if rec.parent_signer_id != owner_rec.signer_id or rec.parent_fingerprint != owner_rec.fingerprint:
fail("MACHINE_PARENT_BINDING_INVALID")
# RegistryOwnerApprovalVerifier handles Owner namespace; generic machine uses registry public key.
if role == "OWNER_AUTHORITY":
verifier = RegistryOwnerApprovalVerifier(self.registry_path, require_root_owner=self.require_root_owner, signer_id=signer_id)
ok = verifier.verify(namespace=namespace, payload=canonical_json_bytes(payload), signature=signature, key_id=key_id)
else:
from aios_single_host_authority import _load_public_key_line
pub = _load_public_key_line(rec.public_key)
ok = verify_sshsig(pub, canonical_json_bytes(payload), signature, namespace=namespace, expected_key_id=key_id)
if not ok:
fail("SIGNED_ARTIFACT_SIGNATURE_INVALID")
return dict(payload)
def owner(self, artifact: SignedOperationalArtifact | Mapping[str, Any]) -> dict[str, Any]:
return self._verify(artifact, role="OWNER_AUTHORITY", signer_id=DEFAULT_OWNER_SIGNER_ID, namespace=OWNER_NAMESPACE)
def machine(self, artifact: SignedOperationalArtifact | Mapping[str, Any]) -> dict[str, Any]:
return self._verify(artifact, role="MACHINE_OPERATIONAL", signer_id=DEFAULT_MACHINE_SIGNER_ID, namespace=MACHINE_NAMESPACE)
@@ -0,0 +1,410 @@
"""Unix-domain fixed-schema service for operational migration authority."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
from pathlib import Path
import socket
import sqlite3
import struct
import stat
import sys
import time
from typing import Any, Mapping
from aios_single_host_authority import (
DEFAULT_REGISTRY_PATH,
OWNER_NAMESPACE,
MACHINE_NAMESPACE,
SingleHostAuthorityError,
)
from aios_single_host_operational_authority import (
OperationalAuthorityError,
OperationalMachineAuthority,
OperationalOwnerAuthority,
OperationalPolicy,
OperationalVerifier,
SignedOperationalArtifact,
verify_binding_scope,
verify_delegation_scope,
)
DEFAULT_POLICY = Path("/etc/aios/single-host-operational-policy.json")
DEFAULT_OWNER_KEY = Path("/var/lib/aios-owner-authority/owner-authority.key")
DEFAULT_MACHINE_KEY = Path("/var/lib/aios-machine-signer/machine-operational.key")
DEFAULT_OWNER_SOCKET = Path("/run/aios/owner-authority.sock")
DEFAULT_MACHINE_SOCKET = Path("/run/aios/machine-signer.sock")
DEFAULT_LEDGER = Path("/var/lib/aios-operational-authority/issuance-ledger.sqlite3")
MAX_REQUEST = 128 * 1024
READINESS_SCHEMA = "aios.signer-readiness-challenge.v1"
class OperationalServiceError(RuntimeError):
def __init__(self, code: str):
super().__init__(code)
self.code = code
def fail(code: str) -> None:
raise OperationalServiceError(code)
def _read_policy(path: Path, *, require_root_owner: bool = True) -> OperationalPolicy:
path=Path(path)
if not path.is_absolute():
fail("OPERATIONAL_POLICY_PATH_INVALID")
flags=os.O_RDONLY
if hasattr(os,"O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
fd=os.open(path,flags)
except OSError as exc:
raise OperationalServiceError("OPERATIONAL_POLICY_UNAVAILABLE") from exc
try:
st=os.fstat(fd)
if not stat.S_ISREG(st.st_mode):
fail("OPERATIONAL_POLICY_NOT_REGULAR")
mode=stat.S_IMODE(st.st_mode)
if mode & 0o022:
fail("OPERATIONAL_POLICY_WRITABLE_BY_GROUP_OR_OTHER")
if require_root_owner and st.st_uid != 0:
fail("OPERATIONAL_POLICY_NOT_ROOT_OWNED")
raw=bytearray()
while len(raw) <= MAX_REQUEST:
chunk=os.read(fd,min(65536,MAX_REQUEST+1-len(raw)))
if not chunk: break
raw.extend(chunk)
if not raw or len(raw)>MAX_REQUEST:
fail("OPERATIONAL_POLICY_UNAVAILABLE")
finally:
os.close(fd)
try:
obj=json.loads(bytes(raw).decode("utf-8"))
except Exception as exc:
raise OperationalServiceError("OPERATIONAL_POLICY_INVALID") from exc
return OperationalPolicy.from_mapping(obj)
def _artifact(value: Mapping[str, Any]) -> SignedOperationalArtifact:
if not isinstance(value, Mapping) or set(value) != {"payload","signature","key_id","namespace"}:
fail("SIGNED_ARTIFACT_INVALID")
if not isinstance(value["payload"], Mapping):
fail("SIGNED_ARTIFACT_INVALID")
return SignedOperationalArtifact(
payload=dict(value["payload"]),
signature=str(value["signature"]),
key_id=str(value["key_id"]),
namespace=str(value["namespace"]),
)
def _peer_uid(conn: socket.socket) -> int:
if not hasattr(socket, "SO_PEERCRED"):
fail("PEER_CREDENTIALS_UNAVAILABLE")
pid, uid, gid = struct.unpack("3i", conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i")))
return uid
class DurableIssuanceLedger:
"""Root-owned exact-idempotent structured signing ledger."""
def __init__(self, path: Path):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
if os.name == "posix": os.chmod(self.path.parent, 0o700)
if self.path.exists() and self.path.is_symlink():
fail("ISSUANCE_LEDGER_SYMLINK_FORBIDDEN")
conn = sqlite3.connect(self.path)
try:
conn.execute("PRAGMA journal_mode=DELETE")
conn.execute("""
CREATE TABLE IF NOT EXISTS issuance (
role TEXT NOT NULL,
operation TEXT NOT NULL,
nonce TEXT NOT NULL,
request_sha256 TEXT NOT NULL,
response_json TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY(role, operation, nonce)
)
""")
conn.commit()
finally:
conn.close()
os.chmod(self.path, 0o600)
def issue(self, *, role: str, operation: str, nonce: str, request: Mapping[str, Any], issuer: Any) -> dict[str, Any]:
request_sha = hashlib.sha256(
json.dumps(request, sort_keys=True, separators=(",",":"), ensure_ascii=False).encode("utf-8")
).hexdigest()
conn = sqlite3.connect(self.path, timeout=5, isolation_level=None)
try:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"SELECT request_sha256,response_json FROM issuance WHERE role=? AND operation=? AND nonce=?",
(role, operation, nonce),
).fetchone()
if row is not None:
if row[0] != request_sha:
conn.rollback()
fail("NONCE_REUSE_WITH_DIFFERENT_REQUEST")
result = json.loads(row[1])
conn.commit()
return result
result = issuer()
encoded = json.dumps(result, sort_keys=True, separators=(",",":"), ensure_ascii=False)
conn.execute(
"INSERT INTO issuance(role,operation,nonce,request_sha256,response_json,created_at) VALUES(?,?,?,?,?,?)",
(role, operation, nonce, request_sha, encoded, int(time.time())),
)
conn.commit()
return result
except sqlite3.IntegrityError:
conn.rollback()
fail("ISSUANCE_LEDGER_RACE")
finally:
conn.close()
class OperationalAuthorityService:
def __init__(
self,
*,
role: str,
policy_path: Path = DEFAULT_POLICY,
registry_path: Path = DEFAULT_REGISTRY_PATH,
owner_key_path: Path = DEFAULT_OWNER_KEY,
machine_key_path: Path = DEFAULT_MACHINE_KEY,
socket_path: Path | None = None,
ledger_path: Path = DEFAULT_LEDGER,
allowed_uids: tuple[int, ...] = (0,),
clock: Any = time.time,
require_root_owner: bool = True,
):
if role not in {"owner","machine"}:
fail("ROLE_INVALID")
self.role = role
self.policy_path = Path(policy_path)
self.registry_path = Path(registry_path)
self.owner_key_path = Path(owner_key_path)
self.machine_key_path = Path(machine_key_path)
self.socket_path = Path(socket_path or (DEFAULT_OWNER_SOCKET if role == "owner" else DEFAULT_MACHINE_SOCKET))
self.allowed_uids = tuple(allowed_uids)
self.ledger = DurableIssuanceLedger(ledger_path)
self.clock = clock
self.require_root_owner = require_root_owner
def _components(self):
policy = _read_policy(self.policy_path, require_root_owner=self.require_root_owner)
verifier = OperationalVerifier(self.registry_path, require_root_owner=self.require_root_owner)
if self.role == "owner":
signer = OperationalOwnerAuthority(self.owner_key_path, policy, require_root_owner=self.require_root_owner, clock=self.clock)
else:
signer = OperationalMachineAuthority(self.machine_key_path, policy, require_root_owner=self.require_root_owner, clock=self.clock)
return policy, verifier, signer
def handle(self, request: Mapping[str, Any], *, peer_uid: int) -> dict[str, Any]:
if peer_uid not in self.allowed_uids:
fail("PEER_UID_FORBIDDEN")
if not isinstance(request, Mapping) or set(request) != {"operation","payload"} or not isinstance(request["payload"], Mapping):
fail("REQUEST_INVALID")
op = request["operation"]
p = request["payload"]
policy, verifier, signer = self._components()
if op == "identity":
return {
"status":"PASS",
"role":self.role,
"policy_digest":policy.digest,
"key_id":signer.fingerprint,
"socket":str(self.socket_path),
"arbitrary_signing":False,
}
if op == "readiness_challenge":
if set(p) != {"challenge_id","challenge_sha256"}:
fail("REQUEST_FIELDS_INVALID")
challenge_id = str(p["challenge_id"])
challenge_sha256 = str(p["challenge_sha256"])
if not challenge_id or len(challenge_id) > 160:
fail("READINESS_CHALLENGE_ID_INVALID")
if not isinstance(challenge_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", challenge_sha256):
fail("READINESS_CHALLENGE_SHA256_INVALID")
namespace = OWNER_NAMESPACE if self.role == "owner" else MACHINE_NAMESPACE
def issue():
art = signer._sign({
"schema": READINESS_SCHEMA,
"role": self.role,
"challenge_id": challenge_id,
"challenge_sha256": challenge_sha256,
"policy_digest": policy.digest,
"issued_at": int(self.clock()),
})
return {"status":"PASS","artifact":art.to_dict()}
return self.ledger.issue(
role=self.role,
operation=op,
nonce=challenge_id,
request=p,
issuer=issue,
)
if self.role == "owner":
if op == "issue_operational_delegation":
if set(p) != {"delegation_id","nonce","ttl_seconds"}:
fail("REQUEST_FIELDS_INVALID")
def issue():
art = signer.issue_operational_delegation(
delegation_id=str(p["delegation_id"]),
nonce=str(p["nonce"]),
ttl_seconds=p["ttl_seconds"],
)
return {"status":"PASS","artifact":art.to_dict()}
return self.ledger.issue(role=self.role, operation=op, nonce=str(p["nonce"]), request=p, issuer=issue)
if op == "issue_expired_task_recovery":
if set(p) != {"delegation","decision_id","nonce","expected_revision","expected_attempt","original_deadline","new_deadline","ttl_seconds"}:
fail("REQUEST_FIELDS_INVALID")
parent_art = _artifact(p["delegation"])
parent = verifier.owner(parent_art)
verify_delegation_scope(parent, policy, now=int(self.clock()))
def issue():
art = signer.issue_recovery_decision(
parent,
decision_id=str(p["decision_id"]),
nonce=str(p["nonce"]),
expected_revision=p["expected_revision"],
expected_attempt=p["expected_attempt"],
original_deadline=p["original_deadline"],
new_deadline=p["new_deadline"],
ttl_seconds=p["ttl_seconds"],
)
return {"status":"PASS","artifact":art.to_dict()}
return self.ledger.issue(role=self.role, operation=op, nonce=str(p["nonce"]), request=p, issuer=issue)
fail("OWNER_OPERATION_FORBIDDEN")
if op == "issue_worker_runtime_binding":
if set(p) != {
"delegation","binding_id","nonce","worker_id","runtime_id","runtime_sha256",
"task_contract_version","supported_contract_versions","required_capabilities",
"worker_capabilities","max_concurrency","ttl_seconds"
}:
fail("REQUEST_FIELDS_INVALID")
parent_art = _artifact(p["delegation"])
parent = verifier.owner(parent_art)
verify_delegation_scope(parent, policy, now=int(self.clock()))
def issue():
art = signer.issue_runtime_binding(
parent,
binding_id=str(p["binding_id"]), nonce=str(p["nonce"]),
worker_id=str(p["worker_id"]), runtime_id=str(p["runtime_id"]),
runtime_sha256=str(p["runtime_sha256"]),
task_contract_version=str(p["task_contract_version"]),
supported_contract_versions=tuple(p["supported_contract_versions"]),
required_capabilities=tuple(p["required_capabilities"]),
worker_capabilities=tuple(p["worker_capabilities"]),
max_concurrency=p["max_concurrency"], ttl_seconds=p["ttl_seconds"],
)
return {"status":"PASS","artifact":art.to_dict()}
return self.ledger.issue(role=self.role, operation=op, nonce=str(p["nonce"]), request=p, issuer=issue)
if op == "issue_worker_runtime_proof":
if set(p) != {
"delegation","binding","proof_id","nonce","heartbeat_at","adapter_readiness",
"resource_readiness","active_claim_count","ttl_seconds"
}:
fail("REQUEST_FIELDS_INVALID")
parent_art = _artifact(p["delegation"])
binding_art = _artifact(p["binding"])
parent = verifier.owner(parent_art)
binding = verifier.machine(binding_art)
verify_delegation_scope(parent, policy, now=int(self.clock()))
verify_binding_scope(binding, parent, policy, now=int(self.clock()))
def issue():
art = signer.issue_runtime_proof(
parent, binding,
proof_id=str(p["proof_id"]), nonce=str(p["nonce"]),
heartbeat_at=p["heartbeat_at"],
adapter_readiness=p["adapter_readiness"],
resource_readiness=p["resource_readiness"],
active_claim_count=p["active_claim_count"],
ttl_seconds=p["ttl_seconds"],
)
return {"status":"PASS","artifact":art.to_dict()}
return self.ledger.issue(role=self.role, operation=op, nonce=str(p["nonce"]), request=p, issuer=issue)
fail("MACHINE_OPERATION_FORBIDDEN")
def serve_forever(self) -> None:
path = self.socket_path
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists() or path.is_symlink():
if path.is_socket():
path.unlink()
else:
fail("SOCKET_PATH_OCCUPIED")
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
sock.bind(str(path))
os.chmod(path, 0o600)
sock.listen(16)
while True:
conn, _ = sock.accept()
with conn:
try:
uid = _peer_uid(conn)
chunks = []
total = 0
while True:
data = conn.recv(65536)
if not data:
break
total += len(data)
if total > MAX_REQUEST:
fail("REQUEST_TOO_LARGE")
chunks.append(data)
if b"\n" in data:
break
raw = b"".join(chunks)
line = raw.split(b"\n",1)[0]
req = json.loads(line.decode("utf-8"))
result = self.handle(req, peer_uid=uid)
out = {"ok":True,"result":result}
except (OperationalServiceError, OperationalAuthorityError, SingleHostAuthorityError) as exc:
out = {"ok":False,"error":getattr(exc,"code",str(exc))}
except Exception:
out = {"ok":False,"error":"INTERNAL_FAIL_CLOSED"}
conn.sendall(json.dumps(out, sort_keys=True, separators=(",",":")).encode("utf-8")+b"\n")
finally:
sock.close()
try:
path.unlink()
except FileNotFoundError:
pass
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--role", choices=("owner","machine"), required=True)
ap.add_argument("--policy", default=str(DEFAULT_POLICY))
ap.add_argument("--registry", default=str(DEFAULT_REGISTRY_PATH))
ap.add_argument("--owner-key", default=str(DEFAULT_OWNER_KEY))
ap.add_argument("--machine-key", default=str(DEFAULT_MACHINE_KEY))
ap.add_argument("--socket")
ns = ap.parse_args()
service = OperationalAuthorityService(
role=ns.role,
policy_path=Path(ns.policy),
registry_path=Path(ns.registry),
owner_key_path=Path(ns.owner_key),
machine_key_path=Path(ns.machine_key),
socket_path=Path(ns.socket) if ns.socket else None,
)
service.serve_forever()
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -14,6 +14,7 @@ import hashlib
import importlib.util
import json
import os
import re
from pathlib import Path
import stat
import time
@@ -285,7 +286,6 @@ class AgentTaskAdmissionPort:
"external_job_id": external_job_id,
}
class TaskAdmissionHostError(RuntimeError):
"""Stable fail-closed host-composition error."""
@@ -839,3 +839,494 @@ def build_task_admission_handlers_from_environment(
)),
post_admission=post_admission,
)
# ---- producer phase core v12r4 (hash-pinned sibling source) ----
import hashlib
import json
import os
from pathlib import Path
import re
import stat
from typing import Any, Mapping
TARGET_TASK_ID="TASK-SINGLEHOST-LINEAGE-SUPERSESSION-P0-20260905"
TARGET_REPOSITORY="aios/aios"
TARGET_BASE_OID="91f104ef62a3b688a9259b96e2ca247e0363b564"
TARGET_BRANCH="agent/singlehost-lineage-supersession-p0-20260905"
TARGET_GENERATION=1
PLAN_SCHEMA_V3="aios.task-admission-execution-plan.v3"
PLAN_FIELDS_V3=frozenset({
"schema","task_id","repository","authority_base_oid","target_branch","generation",
"file_allowlist","file_sha256","capabilities","risk",
})
_SHA256=re.compile(r"^[0-9a-f]{64}$")
_OID=re.compile(r"^[0-9a-f]{40}$")
_JOB_ID=re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,199}$")
class ProducerReceiptError(RuntimeError):
def __init__(self,code:str):
super().__init__(code)
self.code=code
def fail(code:str):
raise ProducerReceiptError(code)
def canonical_json_bytes(value:object)->bytes:
return json.dumps(
value,ensure_ascii=False,sort_keys=True,separators=(",",":"),allow_nan=False
).encode("utf-8")
def _pairs_no_duplicates(pairs):
out={}
for key,value in pairs:
if key in out:
fail("TASK_ADMISSION_EXECUTION_PLAN_DUPLICATE_KEY")
out[key]=value
return out
def strict_json_object(raw:bytes)->dict[str,Any]:
try:
value=json.loads(raw.decode("utf-8"),object_pairs_hook=_pairs_no_duplicates)
except ProducerReceiptError:
raise
except Exception:
fail("TASK_ADMISSION_EXECUTION_PLAN_JSON_INVALID")
if not isinstance(value,dict):
fail("TASK_ADMISSION_EXECUTION_PLAN_INVALID")
if canonical_json_bytes(value)!=raw:
fail("TASK_ADMISSION_EXECUTION_PLAN_NONCANONICAL")
return value
def _gv(grant:object,name:str)->Any:
return grant.get(name) if isinstance(grant,Mapping) else getattr(grant,name,None)
def require_target_grant(grant:object)->None:
expected={
"task_id":TARGET_TASK_ID,
"repository":TARGET_REPOSITORY,
"authority_base_oid":TARGET_BASE_OID,
"target_branch":TARGET_BRANCH,
"generation":TARGET_GENERATION,
}
for name,value in expected.items():
if _gv(grant,name)!=value:
fail("TASK_ADMISSION_POST_OWNER_RESERVATION_SCOPE_INVALID")
digest=_gv(grant,"execution_digest")
if not isinstance(digest,str) or not _SHA256.fullmatch(digest):
fail("TASK_ADMISSION_EXECUTION_DIGEST_INVALID")
def _safe_path(value:object)->str:
if not isinstance(value,str) or not value or "\x00" in value:
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_PATH_INVALID")
p=Path(value)
if p.is_absolute() or ".." in p.parts or "." in p.parts or p.as_posix()!=value:
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_PATH_INVALID")
if value==".git" or value.startswith(".git/"):
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_PATH_INVALID")
return value
def _unique_strings(value:object,code:str)->list[str]:
if (
not isinstance(value,list) or not value
or not all(isinstance(x,str) and x for x in value)
or len(set(value))!=len(value)
):
fail(code)
return list(value)
def validate_v3_plan(*,grant:object,raw:bytes)->dict[str,Any]:
require_target_grant(grant)
plan=strict_json_object(raw)
if set(plan)!=PLAN_FIELDS_V3:
fail("TASK_ADMISSION_EXECUTION_PLAN_FIELDS_INVALID")
if plan["schema"]!=PLAN_SCHEMA_V3:
fail("TASK_ADMISSION_EXECUTION_PLAN_SCHEMA_INVALID")
fields=(
("task_id","TASK_ADMISSION_EXECUTION_PLAN_TASK_MISMATCH"),
("repository","TASK_ADMISSION_EXECUTION_PLAN_REPOSITORY_MISMATCH"),
("authority_base_oid","TASK_ADMISSION_EXECUTION_PLAN_BASE_MISMATCH"),
("target_branch","TASK_ADMISSION_EXECUTION_PLAN_BRANCH_MISMATCH"),
("generation","TASK_ADMISSION_EXECUTION_PLAN_GENERATION_MISMATCH"),
)
for field,code in fields:
if plan[field]!=_gv(grant,field):
fail(code)
if not _OID.fullmatch(str(plan["authority_base_oid"])):
fail("TASK_ADMISSION_EXECUTION_PLAN_BASE_INVALID")
allow=[_safe_path(x) for x in _unique_strings(
plan["file_allowlist"],"TASK_ADMISSION_EXECUTION_PLAN_ALLOWLIST_INVALID"
)]
if allow!=list(_gv(grant,"file_allowlist") or ()):
fail("TASK_ADMISSION_EXECUTION_PLAN_ALLOWLIST_MISMATCH")
hashes=plan["file_sha256"]
if not isinstance(hashes,dict) or set(hashes)!=set(allow):
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_HASH_SET_MISMATCH")
if any(not isinstance(v,str) or not _SHA256.fullmatch(v) for v in hashes.values()):
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_HASH_INVALID")
caps=_unique_strings(
plan["capabilities"],"TASK_ADMISSION_EXECUTION_PLAN_CAPABILITIES_INVALID"
)
if caps!=list(_gv(grant,"capabilities") or ()):
fail("TASK_ADMISSION_EXECUTION_PLAN_CAPABILITIES_MISMATCH")
if plan["risk"]!=_gv(grant,"risk"):
fail("TASK_ADMISSION_EXECUTION_PLAN_RISK_MISMATCH")
if hashlib.sha256(raw).hexdigest()!=_gv(grant,"execution_digest"):
fail("TASK_ADMISSION_EXECUTION_PLAN_DIGEST_MISMATCH")
return plan
def load_v3_plan(root:Path,grant:object)->dict[str,Any]:
require_target_grant(grant)
try:
rst=root.lstat()
except OSError:
fail("TASK_ADMISSION_EXECUTION_PLAN_UNAVAILABLE")
if (
stat.S_ISLNK(rst.st_mode) or not stat.S_ISDIR(rst.st_mode)
or rst.st_uid!=0 or rst.st_mode&0o022
):
fail("TASK_ADMISSION_EXECUTION_PLAN_DIRECTORY_INVALID")
path=root/(str(_gv(grant,"execution_digest"))+".json")
try:
st=path.lstat()
except OSError:
fail("TASK_ADMISSION_EXECUTION_PLAN_UNAVAILABLE")
if (
stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode)
or st.st_uid!=0 or st.st_mode&0o022
):
fail("TASK_ADMISSION_EXECUTION_PLAN_FILE_INVALID")
try:
raw=path.read_bytes()
except OSError:
fail("TASK_ADMISSION_EXECUTION_PLAN_UNAVAILABLE")
return validate_v3_plan(grant=grant,raw=raw)
def reservation_idempotency_key(grant:object)->str:
require_target_grant(grant)
material=(
str(_gv(grant,"grant_id"))+"\0"+str(_gv(grant,"execution_digest"))
).encode("utf-8")
return "task-admission-execution-reservation-"+hashlib.sha256(material).hexdigest()[:32]
def build_reservation_request(grant:object)->dict[str,Any]:
require_target_grant(grant)
return {
"task_id":TARGET_TASK_ID,
"repository":TARGET_REPOSITORY,
"base_oid":TARGET_BASE_OID,
"target_branch":TARGET_BRANCH,
"generation":TARGET_GENERATION,
"risk":str(_gv(grant,"risk")),
"description":"Reserve one inert durable GitOps Job identity after exact Owner TaskAdmission authorization; no Agent Task claim or execution.",
"idempotency_key":reservation_idempotency_key(grant),
}
def _mapping(value:object,code:str)->Mapping[str,Any]:
if not isinstance(value,Mapping):
fail(code)
return value
def validate_bootstrap_response(
*,
grant:object,
request:Mapping[str,Any],
response:object,
expected_side_effect_state:str,
)->str:
require_target_grant(grant)
result=_mapping(response,"TASK_ADMISSION_EXECUTION_RESERVATION_RESPONSE_INVALID")
job_id=result.get("job_id")
if not isinstance(job_id,str) or not _JOB_ID.fullmatch(job_id):
fail("TASK_ADMISSION_EXECUTION_RESERVATION_JOB_ID_INVALID")
expected={
"task_id":TARGET_TASK_ID,
"job_id":job_id,
"repository":TARGET_REPOSITORY,
"base_oid":TARGET_BASE_OID,
"expected_branch_oid":TARGET_BASE_OID,
"target_branch":TARGET_BRANCH,
"branch_mode":"CONTINUE_EXISTING",
"generation":TARGET_GENERATION,
"state":"BOUND",
}
for key,value in expected.items():
if result.get(key)!=value:
fail("TASK_ADMISSION_EXECUTION_RESERVATION_BINDING_CONFLICT")
receipt=_mapping(
result.get("reservation_receipt"),
"TASK_ADMISSION_EXECUTION_RESERVATION_RECEIPT_MISSING",
)
expected_keys={
"job_id","state","side_effect_state",
"producer_idempotency_key","request_idempotency_key",
}
if set(receipt)!=expected_keys:
fail("TASK_ADMISSION_EXECUTION_RESERVATION_RECEIPT_FIELDS_INVALID")
producer_key=receipt.get("producer_idempotency_key")
if (
receipt.get("job_id")!=job_id
or receipt.get("state")!="REQUESTED"
or receipt.get("side_effect_state")!=expected_side_effect_state
or not isinstance(producer_key,str)
or not _SHA256.fullmatch(producer_key)
or job_id!="job-"+producer_key[:32]
or receipt.get("request_idempotency_key")!=request.get("idempotency_key")
):
fail("TASK_ADMISSION_EXECUTION_RESERVATION_RECEIPT_CONFLICT")
return job_id
def _approved_binding_fingerprint(response:Mapping[str,Any])->str:
keys=(
"task_id","job_id","repository","base_oid","expected_branch_oid",
"target_branch","branch_mode","generation","state","reservation_receipt",
)
return hashlib.sha256(canonical_json_bytes({k:response.get(k) for k in keys})).hexdigest()
def _stable_receipt_identity(response:Mapping[str,Any])->dict[str,Any]:
receipt=_mapping(response.get("reservation_receipt"),"TASK_ADMISSION_EXECUTION_RESERVATION_RECEIPT_MISSING")
return {
"task_id":response.get("task_id"),
"job_id":response.get("job_id"),
"repository":response.get("repository"),
"base_oid":response.get("base_oid"),
"expected_branch_oid":response.get("expected_branch_oid"),
"target_branch":response.get("target_branch"),
"branch_mode":response.get("branch_mode"),
"generation":response.get("generation"),
"binding_state":response.get("state"),
"receipt_state":receipt.get("state"),
"producer_idempotency_key":receipt.get("producer_idempotency_key"),
"request_idempotency_key":receipt.get("request_idempotency_key"),
}
def validate_bootstrap_replay(
*,
grant:object,
request:Mapping[str,Any],
first:object,
replay:object,
replay_side_effect_state:str,
)->str:
first_id=validate_bootstrap_response(
grant=grant,request=request,response=first,expected_side_effect_state="QUEUED"
)
replay_id=validate_bootstrap_response(
grant=grant,request=request,response=replay,
expected_side_effect_state=replay_side_effect_state
)
if first_id!=replay_id:
fail("TASK_ADMISSION_EXECUTION_RESERVATION_REPLAY_JOB_CONFLICT")
first_map=_mapping(first,"TASK_ADMISSION_EXECUTION_RESERVATION_RESPONSE_INVALID")
replay_map=_mapping(replay,"TASK_ADMISSION_EXECUTION_RESERVATION_RESPONSE_INVALID")
if _stable_receipt_identity(first_map)!=_stable_receipt_identity(replay_map):
fail("TASK_ADMISSION_EXECUTION_RESERVATION_REPLAY_IDENTITY_CONFLICT")
return first_id
# ===== AIOS FORMAL LINEAGE POST-OWNER PRODUCER NATIVE V12 — BEGIN =====
import inspect as _fl_inspect
_FL_HARDENING_PROVIDER = ExecutionPlanDirectoryProvider
_FL_BUILD_FROM_ACTIVATION = build_task_admission_handlers_from_activation_file
_FL_BUILD_FROM_ENVIRONMENT = globals().get("build_task_admission_handlers_from_environment")
_FL_APPROVED_BOOTSTRAP_SHA256="758895a8407568890fd2f8eb2f232a3dbf2cfa369e5f65ae8c6975bf8d01e3b5"
REPLAY_SIDE_EFFECT_STATE="SAME_JOB"
def _fl_error(code:str,status:int=403):
cls=globals().get("TaskAdmissionError",RuntimeError)
try:
return cls(code,status)
except TypeError:
return cls(code)
def _fl_sha(path:Path)->str:
h=hashlib.sha256()
with path.open("rb") as f:
for chunk in iter(lambda:f.read(1024*1024),b""):
h.update(chunk)
return h.hexdigest()
def _fl_provider_root(provider:object)->Path:
found=[]
for name in ("root","directory","plan_dir","execution_plan_dir"):
raw=getattr(provider,name,None)
if isinstance(raw,(str,os.PathLike)):
p=Path(raw)
if p.is_absolute():
found.append(p)
unique={str(p) for p in found}
if len(unique)!=1:
raise _fl_error("TASK_ADMISSION_EXECUTION_PLAN_PROVIDER_ROOT_INVALID",503)
return Path(next(iter(unique)))
def _fl_assert_bootstrap_source(bootstrap:object)->None:
raw=_fl_inspect.getsourcefile(type(bootstrap))
if not raw:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_BOOTSTRAP_SOURCE_UNKNOWN",503)
path=Path(raw)
if not path.is_file() or path.is_symlink():
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_BOOTSTRAP_SOURCE_INVALID",503)
if _fl_sha(path)!=_FL_APPROVED_BOOTSTRAP_SHA256:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_BOOTSTRAP_SOURCE_DRIFT",503)
class _FLProducerReceiptReserver:
def __init__(self,runtime:object):
self.runtime=runtime
def __call__(self,grant:object,plan:Mapping[str,Any])->str:
authority=getattr(self.runtime,"authority",None)
bootstrap=getattr(self.runtime,"bootstrap",None)
bindings=getattr(self.runtime,"bindings",None)
if (
authority is None or bootstrap is None
or not callable(getattr(bootstrap,"create",None))
or not isinstance(bindings,dict)
):
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_BACKEND_INVALID",503)
_fl_assert_bootstrap_source(bootstrap)
try:
main_before=authority.read_ref("refs/heads/main")
branch_before=authority.read_ref("refs/heads/"+TARGET_BRANCH)
except Exception as exc:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_AUTHORITY_UNAVAILABLE",503) from exc
if main_before!=TARGET_BASE_OID or branch_before!=TARGET_BASE_OID:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_AUTHORITY_DRIFT",409)
request=build_reservation_request(grant)
request_before=dict(request)
try:
first=bootstrap.create(request)
if request!=request_before:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_REQUEST_MUTATED",503)
validate_bootstrap_response(grant=grant,request=request,response=first,expected_side_effect_state="QUEUED")
replay=bootstrap.create(request)
if request!=request_before:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_REQUEST_MUTATED",503)
job_id=validate_bootstrap_replay(
grant=grant,request=request,first=first,replay=replay,
replay_side_effect_state=REPLAY_SIDE_EFFECT_STATE,
)
except ProducerReceiptError as exc:
raise _fl_error(exc.code,409) from exc
except Exception as exc:
task_error=globals().get("TaskAdmissionError")
if isinstance(task_error,type) and isinstance(exc,task_error):
raise
raise _fl_error(
str(exc) or "TASK_ADMISSION_EXECUTION_RESERVATION_FAILED",503
) from exc
# Existing runtime's public createGovernedTask projection performs this
# in-memory binding update after bootstrap.create(). Mirror it only
# after the exact producer receipt + replay proof has passed.
bindings[TARGET_TASK_ID]=dict(replay)
try:
main_after=authority.read_ref("refs/heads/main")
branch_after=authority.read_ref("refs/heads/"+TARGET_BRANCH)
except Exception as exc:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_AUTHORITY_UNAVAILABLE",503) from exc
if main_after!=main_before or branch_after!=branch_before:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_REF_MUTATION",409)
return job_id
class V3ProducerReceiptExecutionPlanDirectoryProvider(_FL_HARDENING_PROVIDER):
_reserver:_FLProducerReceiptReserver|None=None
def _target(self,grant:object)->Mapping[str,Any]:
if self._reserver is None:
raise _fl_error("TASK_ADMISSION_EXECUTION_RESERVATION_BACKEND_INVALID",503)
try:
plan=load_v3_plan(_fl_provider_root(self),grant)
job_id=self._reserver(grant,plan)
except ProducerReceiptError as exc:
raise _fl_error(exc.code,403) from exc
return {**plan,"external_job_id":job_id}
def __call__(self,grant:object)->Mapping[str,Any]:
if getattr(grant,"task_id",None)!=TARGET_TASK_ID:
parent=getattr(super(),"__call__",None)
if callable(parent):
return parent(grant)
raise _fl_error("TASK_ADMISSION_EXECUTION_PLAN_PROVIDER_CONTRACT_INVALID",503)
return self._target(grant)
def load(self,grant:object)->Mapping[str,Any]:
if getattr(grant,"task_id",None)!=TARGET_TASK_ID:
parent=getattr(super(),"load",None)
if callable(parent):
return parent(grant)
parent_call=getattr(super(),"__call__",None)
if callable(parent_call):
return parent_call(grant)
raise _fl_error("TASK_ADMISSION_EXECUTION_PLAN_PROVIDER_CONTRACT_INVALID",503)
return self._target(grant)
def _fl_runtime_from_post_admission(post_admission:object)->object:
runtime=getattr(post_admission,"__self__",None)
if runtime is None or not callable(getattr(runtime,"project_admitted_task",None)):
raise RuntimeError("TASK_ADMISSION_POST_OWNER_RUNTIME_BINDING_INVALID")
return runtime
def _fl_call(builder,*args,post_admission=None,**kwargs):
if post_admission is None:
raise RuntimeError("TASK_ADMISSION_POST_OWNER_POST_ADMISSION_REQUIRED")
reserver=_FLProducerReceiptReserver(_fl_runtime_from_post_admission(post_admission))
class BoundProvider(V3ProducerReceiptExecutionPlanDirectoryProvider):
_reserver=reserver
global ExecutionPlanDirectoryProvider
previous=ExecutionPlanDirectoryProvider
ExecutionPlanDirectoryProvider=BoundProvider
try:
return builder(*args,post_admission=post_admission,**kwargs)
finally:
ExecutionPlanDirectoryProvider=previous
def build_task_admission_handlers_from_activation_file(
activation_file:Path,*,
caller_service:str="aios-chatgpt-bridge-v2-staging.service",
post_admission=None,
):
return _fl_call(
_FL_BUILD_FROM_ACTIVATION,activation_file,
caller_service=caller_service,post_admission=post_admission,
)
def build_task_admission_handlers_from_environment(
*,
caller_service:str="aios-chatgpt-bridge-v2-staging.service",
post_admission=None,
):
if not callable(_FL_BUILD_FROM_ENVIRONMENT):
raise RuntimeError("TASK_ADMISSION_ENVIRONMENT_BUILDER_MISSING")
return _fl_call(
_FL_BUILD_FROM_ENVIRONMENT,
caller_service=caller_service,post_admission=post_admission,
)
ExecutionPlanDirectoryProvider=V3ProducerReceiptExecutionPlanDirectoryProvider
# ===== AIOS FORMAL LINEAGE POST-OWNER PRODUCER NATIVE V12 — END =====
@@ -0,0 +1,394 @@
"""Append-only ClaimGuard binding for the superseded Single-Host child Task."""
from __future__ import annotations
from datetime import datetime, timezone
import hashlib
import json
import re
import sqlite3
from typing import Any, Callable, Mapping
from aios_task_lineage_supersession_initial_projection import (
PROJECTION_RECEIPT_SCHEMA,
PROJECTION_TABLE,
SOURCE_BUSINESS_EXECUTION_DIGEST,
)
SOURCE_TASK_ID = "TASK-SINGLEHOST-TRUSTROOT-P0-20260904"
CHILD_TASK_ID = SOURCE_TASK_ID + "-S1"
WORKER_ID = "p0-task-principal-v15-20260830"
REPOSITORY = "aios/aios"
TARGET_BRANCH = "agent/single-host-trust-root-p0-20260904"
GENERATION = 1
EXTERNAL_JOB_ID = "job-4e10f92a6a5742518a9f1d8745c845ab"
BINDING_SCHEMA = "aios.task-claim-guard-binding.v2"
RECEIPT_SCHEMA = "aios.task-claim-guard-binding-receipt.v2"
HEX64 = re.compile(r"^[0-9a-f]{64}$")
class ClaimGuardBindingError(RuntimeError):
def __init__(self, code: str, http_status: int = 409):
super().__init__(code)
self.code = code
self.http_status = int(http_status)
def _fail(code: str, status: int = 409) -> None:
raise ClaimGuardBindingError(code, status)
def canonical_json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
def _sha_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _sha_json(value: Any) -> str:
return _sha_text(canonical_json(value))
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def _row_dict(row: Any) -> dict[str, Any]:
if row is None:
return {}
if hasattr(row, "keys"):
return {str(key): row[key] for key in row.keys()}
_fail("TASK_ROW_MAPPING_UNAVAILABLE", 500)
def _json(raw: Any, code: str) -> Any:
if not isinstance(raw, str):
_fail(code)
try:
return json.loads(raw)
except (TypeError, ValueError) as exc:
raise ClaimGuardBindingError(code) from exc
def _default_policy_factory(value: Mapping[str, Any]):
try:
from aios_single_host_operational_authority import OperationalPolicy
except ImportError as exc:
raise ClaimGuardBindingError("OPERATIONAL_POLICY_RUNTIME_UNAVAILABLE", 503) from exc
return OperationalPolicy.from_mapping(value)
def _policy_value(policy: Any, name: str) -> Any:
try:
return getattr(policy, name)
except AttributeError as exc:
raise ClaimGuardBindingError("OPERATIONAL_POLICY_FIELDS_INVALID") from exc
def _load_projection_evidence(connection, relation) -> dict[str, Any]:
try:
row = connection.execute(
f"SELECT * FROM {PROJECTION_TABLE} WHERE task_id=? AND source_task_id=? "
"AND supersession_ordinal=?",
(CHILD_TASK_ID, SOURCE_TASK_ID, 1),
).fetchone()
except sqlite3.OperationalError as exc:
raise ClaimGuardBindingError(
"FORMAL_CHILD_INITIAL_PROJECTION_SCHEMA_UNAVAILABLE", 503
) from exc
if row is None:
_fail("FORMAL_CHILD_INITIAL_PROJECTION_REQUIRED")
value = _row_dict(row)
receipt = _json(value.get("receipt_json"), "CHILD_PROJECTION_RECEIPT_INVALID")
canonical_digest = value.get("canonical_child_execution_digest")
source_digest = value.get("source_business_execution_digest")
exact = {
"task_id": CHILD_TASK_ID,
"source_task_id": SOURCE_TASK_ID,
"supersession_ordinal": 1,
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_id": EXTERNAL_JOB_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"owner_decision_sha256": relation["owner_decision_sha256"],
}
if any(value.get(key) != expected for key, expected in exact.items()):
_fail("CHILD_PROJECTION_LINEAGE_MISMATCH")
if (
not isinstance(canonical_digest, str)
or HEX64.fullmatch(canonical_digest) is None
or canonical_digest == SOURCE_BUSINESS_EXECUTION_DIGEST
or source_digest != SOURCE_BUSINESS_EXECUTION_DIGEST
or _sha_text(value["receipt_json"]) != value.get("receipt_sha256")
or not isinstance(receipt, Mapping)
or receipt.get("schema") != PROJECTION_RECEIPT_SCHEMA
or receipt.get("canonical_child_execution_digest") != canonical_digest
or receipt.get("source_business_execution_digest") != source_digest
or receipt.get("owner_supersession_decision_sha256")
!= relation["owner_decision_sha256"]
):
_fail("CHILD_PROJECTION_RECEIPT_MISMATCH")
return {
**value,
"canonical_child_execution_digest": canonical_digest,
"source_business_execution_digest": source_digest,
}
def install_child_binding(
connection,
*,
policy: Mapping[str, Any],
runtime_id: str,
runtime_observation_digest: str,
task_contract_version: str,
policy_factory: Callable[[Mapping[str, Any]], Any] = _default_policy_factory,
now: str | None = None,
) -> dict[str, Any]:
"""Generate and persist the only permitted child binding from current DB state."""
if not isinstance(policy, Mapping):
_fail("OPERATIONAL_POLICY_INVALID")
if not isinstance(runtime_id, str) or not runtime_id or len(runtime_id) > 512:
_fail("RUNTIME_ID_INVALID")
if not isinstance(runtime_observation_digest, str) or HEX64.fullmatch(runtime_observation_digest) is None:
_fail("RUNTIME_OBSERVATION_DIGEST_INVALID")
if not isinstance(task_contract_version, str) or not task_contract_version or len(task_contract_version) > 128:
_fail("TASK_CONTRACT_VERSION_INVALID")
relation = connection.execute(
"SELECT * FROM task_lineage_supersessions WHERE source_task_id=? AND supersession_ordinal=1 AND child_task_id=?",
(SOURCE_TASK_ID, CHILD_TASK_ID),
).fetchone()
if relation is None:
_fail("FORMAL_SUPERSESSION_REQUIRED")
projection = _load_projection_evidence(connection, relation)
canonical_execution_digest = projection["canonical_child_execution_digest"]
task_row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (CHILD_TASK_ID,)).fetchone()
if task_row is None:
_fail("CHILD_TASK_NOT_FOUND", 404)
task = _row_dict(task_row)
if (
task.get("status") != "QUEUED" or int(task.get("revision", -1)) < 2
or int(task.get("attempt", -1)) != 0 or task.get("claimed_by") is not None
or task.get("lease_token_hash") is not None or task.get("lease_expires_at") is not None
or task.get("result_json") is not None or task.get("verification_json") is not None
):
_fail("CHILD_PROJECTED_CLAIMABLE_STATE_REQUIRED")
event_rows = connection.execute(
"SELECT event_type,payload_json FROM events WHERE task_id=? ORDER BY event_id",
(CHILD_TASK_ID,),
).fetchall()
if [row[0] for row in event_rows] != [
"TASK_CREATED",
"TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED",
]:
_fail("CANONICAL_INITIAL_PROJECTION_REQUIRED")
projection_event = _json(
event_rows[-1][1], "CANONICAL_INITIAL_PROJECTION_EVENT_INVALID"
)
if (
projection_event.get("source_business_execution_digest")
!= SOURCE_BUSINESS_EXECUTION_DIGEST
or projection_event.get("canonical_child_execution_digest")
!= canonical_execution_digest
or projection_event.get("owner_supersession_decision_sha256")
!= relation["owner_decision_sha256"]
or _sha_json(projection_event) != projection["projection_event_sha256"]
):
_fail("CANONICAL_INITIAL_PROJECTION_EVENT_MISMATCH")
metadata_raw = task.get("metadata_json")
capabilities_raw = task.get("required_capabilities_json")
metadata = _json(metadata_raw, "CHILD_METADATA_INVALID")
capabilities = _json(capabilities_raw, "CHILD_CAPABILITIES_INVALID")
if (
_sha_text(metadata_raw) != projection["metadata_sha256"]
or _sha_text(capabilities_raw)
!= projection["required_capabilities_sha256"]
):
_fail("CHILD_PROJECTION_TASK_DIGEST_MISMATCH")
if not isinstance(metadata, Mapping) or not isinstance(capabilities, list) or not capabilities:
_fail("CHILD_PROJECTION_INVALID")
if any(not isinstance(value, str) or not value for value in capabilities) or len(set(capabilities)) != len(capabilities):
_fail("CHILD_CAPABILITIES_INVALID")
execution = metadata.get("execution")
if not isinstance(execution, Mapping):
_fail("CHILD_EXECUTION_PROJECTION_MISSING")
exact_execution = {
"task_id": CHILD_TASK_ID, "repository": REPOSITORY,
"target_branch": TARGET_BRANCH, "generation": GENERATION,
"execution_digest": canonical_execution_digest, "job_id": EXTERNAL_JOB_ID,
}
if any(execution.get(key) != value for key, value in exact_execution.items()):
_fail("CHILD_EXECUTION_BINDING_MISMATCH")
parsed_policy = policy_factory(dict(policy))
if hasattr(parsed_policy, "validate"):
parsed_policy.validate()
policy_exact = {
"task_id": CHILD_TASK_ID, "repository": REPOSITORY,
"target_branch": TARGET_BRANCH, "generation": GENERATION,
"execution_plan_sha256": canonical_execution_digest,
"external_job_identity": EXTERNAL_JOB_ID,
}
if any(_policy_value(parsed_policy, key) != value for key, value in policy_exact.items()):
_fail("OPERATIONAL_POLICY_SCOPE_MISMATCH")
if tuple(_policy_value(parsed_policy, "allowed_worker_ids")) != (WORKER_ID,):
_fail("OPERATIONAL_POLICY_WORKER_MISMATCH")
expected_oid = execution.get("expected_branch_oid", execution.get("base_oid"))
if expected_oid is not None and _policy_value(parsed_policy, "base_oid") != expected_oid:
_fail("OPERATIONAL_POLICY_BASE_OID_MISMATCH")
policy_json = canonical_json(dict(policy))
policy_sha = _sha_text(policy_json)
policy_digest = str(_policy_value(parsed_policy, "digest"))
if policy_digest != policy_sha:
_fail("OPERATIONAL_POLICY_DIGEST_MISMATCH")
binding = {
"schema": BINDING_SCHEMA, "task_id": CHILD_TASK_ID,
"source_task_id": SOURCE_TASK_ID, "supersession_ordinal": 1,
"expected_revision": int(task["revision"]),
"metadata_sha256": _sha_text(metadata_raw),
"required_capabilities_sha256": _sha_json(sorted(capabilities)),
"worker_id": WORKER_ID,
"execution_digest": canonical_execution_digest,
"canonical_child_execution_digest": canonical_execution_digest,
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"repository": REPOSITORY, "target_branch": TARGET_BRANCH, "generation": GENERATION,
"external_job_id": EXTERNAL_JOB_ID, "policy_sha256": policy_sha,
"policy_digest": policy_digest, "runtime_id": runtime_id,
"runtime_observation_digest": runtime_observation_digest,
"task_contract_version": task_contract_version,
"owner_supersession_decision_sha256": relation["owner_decision_sha256"],
}
binding_json = canonical_json(binding)
binding_sha = _sha_text(binding_json)
binding_id = "sha256:" + binding_sha
existing = connection.execute(
"SELECT binding_sha256 FROM task_claim_guard_bindings_v2 WHERE task_id=?", (CHILD_TASK_ID,)
).fetchone()
if existing is not None:
if existing[0] != binding_sha:
_fail("CLAIM_GUARD_BINDING_ALREADY_EXISTS")
receipt = connection.execute(
"SELECT receipt_json FROM task_claim_guard_binding_receipts_v2 WHERE binding_id=?", (binding_id,)
).fetchone()
if receipt is None:
_fail("CLAIM_GUARD_BINDING_RECEIPT_MISSING", 500)
return {**json.loads(receipt[0]), "idempotent": True}
created_at = now or _now()
connection.execute(
"""INSERT INTO task_claim_guard_bindings_v2(
binding_id,task_id,source_task_id,supersession_ordinal,expected_revision,
metadata_sha256,required_capabilities_sha256,worker_id,execution_digest,
repository,target_branch,generation,policy_json,policy_sha256,binding_json,
binding_sha256,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(binding_id, CHILD_TASK_ID, SOURCE_TASK_ID, 1, binding["expected_revision"],
binding["metadata_sha256"], binding["required_capabilities_sha256"], WORKER_ID,
canonical_execution_digest, REPOSITORY, TARGET_BRANCH, GENERATION, policy_json, policy_sha,
binding_json, binding_sha, created_at),
)
receipt = {
"schema": RECEIPT_SCHEMA, "status": "BOUND", "binding_id": binding_id,
"binding_sha256": binding_sha, "task_id": CHILD_TASK_ID,
"expected_revision": binding["expected_revision"],
"metadata_sha256": binding["metadata_sha256"], "worker_id": WORKER_ID,
"execution_digest": canonical_execution_digest,
"canonical_child_execution_digest": canonical_execution_digest,
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH, "generation": GENERATION,
"owner_supersession_decision_sha256": relation["owner_decision_sha256"],
}
receipt_sha = _sha_json(receipt)
receipt_id = "sha256:" + receipt_sha
connection.execute(
"INSERT INTO task_claim_guard_binding_receipts_v2 VALUES(?,?,?,?,?,?)",
(receipt_id, binding_id, CHILD_TASK_ID, canonical_json(receipt), receipt_sha, created_at),
)
return {**receipt, "receipt_id": receipt_id, "receipt_sha256": receipt_sha, "idempotent": False}
def _load_child_binding(connection) -> tuple[dict[str, Any], dict[str, Any]]:
row = connection.execute(
"SELECT * FROM task_claim_guard_bindings_v2 WHERE task_id=?", (CHILD_TASK_ID,)
).fetchone()
if row is None:
_fail("CHILD_CLAIM_GUARD_BINDING_MISSING")
value = _row_dict(row)
binding = _json(value["binding_json"], "CLAIM_GUARD_BINDING_INVALID")
policy = _json(value["policy_json"], "OPERATIONAL_POLICY_INVALID")
relation = connection.execute(
"SELECT * FROM task_lineage_supersessions WHERE source_task_id=? "
"AND supersession_ordinal=1 AND child_task_id=?",
(SOURCE_TASK_ID, CHILD_TASK_ID),
).fetchone()
if relation is None:
_fail("FORMAL_SUPERSESSION_REQUIRED")
projection = _load_projection_evidence(connection, relation)
canonical_execution_digest = projection["canonical_child_execution_digest"]
if (
_sha_text(value["binding_json"]) != value["binding_sha256"]
or binding.get("schema") != BINDING_SCHEMA
or value.get("execution_digest") != canonical_execution_digest
or binding.get("execution_digest") != canonical_execution_digest
or binding.get("canonical_child_execution_digest")
!= canonical_execution_digest
or binding.get("source_business_execution_digest")
!= SOURCE_BUSINESS_EXECUTION_DIGEST
or binding.get("owner_supersession_decision_sha256")
!= relation["owner_decision_sha256"]
or policy.get("execution_plan_sha256") != canonical_execution_digest
):
_fail("CLAIM_GUARD_BINDING_DIGEST_MISMATCH")
task = connection.execute("SELECT * FROM tasks WHERE task_id=?", (CHILD_TASK_ID,)).fetchone()
task = _row_dict(task)
if (
task.get("status") != "QUEUED" or int(task.get("revision", -1)) != int(binding["expected_revision"])
or int(task.get("attempt", -1)) != 0 or task.get("claimed_by") is not None
or task.get("lease_token_hash") is not None or task.get("lease_expires_at") is not None
or _sha_text(task.get("metadata_json", "")) != binding["metadata_sha256"]
or task.get("result_json") is not None or task.get("verification_json") is not None
):
_fail("CHILD_CLAIM_GUARD_TASK_CAS_MISMATCH")
return binding, policy
def _run_child_guard(connection, binding, policy_value, agent_id):
try:
from aios_single_host_operational_authority import OperationalPolicy, OperationalVerifier
from aios_singlehost_claim_guard_stage1 import _LiveReader, REGISTRY_PATH
from aios_task_claim_guard import ClaimGuard, ClaimGuardError as CoreError
except ImportError as exc:
raise ClaimGuardBindingError("CLAIM_GUARD_RUNTIME_UNAVAILABLE", 503) from exc
policy = OperationalPolicy.from_mapping(policy_value)
cfg = {
"task_id": CHILD_TASK_ID, "expected_claim_revision": binding["expected_revision"],
"task_metadata_sha256": binding["metadata_sha256"],
"required_capabilities_sha256": binding["required_capabilities_sha256"],
"task_contract_version": binding["task_contract_version"], "worker_id": WORKER_ID,
"policy_digest": binding["policy_digest"], "runtime_id": binding["runtime_id"],
"runtime_observation_digest": binding["runtime_observation_digest"],
}
reader = _LiveReader(connection, policy, cfg, now=int(datetime.now(timezone.utc).timestamp()))
try:
ClaimGuard(policy, OperationalVerifier(REGISTRY_PATH, require_root_owner=True)).validate_in_transaction(
reader, task_id=CHILD_TASK_ID,
expected_revision=int(binding["expected_revision"]), worker_id=agent_id,
)
except CoreError as exc:
raise ClaimGuardBindingError(exc.code) from exc
def enforce_claim_guard(connection, *, task_id: str, agent_id: str) -> None:
"""Exact two-lineage dispatcher; unrelated Tasks retain their prior behavior."""
if task_id == SOURCE_TASK_ID:
try:
from aios_singlehost_claim_guard_stage1 import enforce_claim_guard as frozen_guard
except ImportError as exc:
raise ClaimGuardBindingError("FROZEN_SOURCE_CLAIM_GUARD_UNAVAILABLE", 503) from exc
frozen_guard(connection, task_id=task_id, agent_id=agent_id)
return
if task_id != CHILD_TASK_ID:
return
if agent_id != WORKER_ID:
_fail("WORKER_ID_MISMATCH", 403)
binding, policy = _load_child_binding(connection)
_run_child_guard(connection, binding, policy, agent_id)
@@ -0,0 +1,339 @@
"""One-time, Owner-signed, append-only supersession for one exact Task lineage."""
from __future__ import annotations
from contextlib import closing
from datetime import datetime, timedelta, timezone
import hashlib
import json
from pathlib import Path
import re
import sqlite3
from typing import Any, Callable, Mapping, Protocol
DECISION_SCHEMA = "aios.task-lineage-supersession-decision.v1"
RECEIPT_SCHEMA = "aios.task-lineage-supersession-receipt.v1"
OWNER_NAMESPACE = "aios-owner-approval-v1"
OWNER_PRINCIPAL = "aios-task-grant-signer"
OWNER_KEY_ID = "SHA256:rPqpxfPHtaZPM3zSWojX8Q7VxzCknsYqOokyo/aAM3A"
SOURCE_TASK_ID = "TASK-SINGLEHOST-TRUSTROOT-P0-20260904"
CHILD_TASK_ID = SOURCE_TASK_ID + "-S1"
REPOSITORY = "aios/aios"
TARGET_BRANCH = "agent/single-host-trust-root-p0-20260904"
GENERATION = 1
EXECUTION_DIGEST = "e99bda6a251832eb3865271eb5687af7eea44dd261144cdcdfa57104c0780536"
EXTERNAL_JOB_ID = "job-4e10f92a6a5742518a9f1d8745c845ab"
SUPERSESSION_ORDINAL = 1
CANONICAL_PROFILE = "aios.task-initial-execution.v1"
CANONICAL_PROFILE_VERSION = 1
SUPERSESSION_REASON = "formal-lineage-supersession-after-expired-unclaimed-source"
MAX_TTL = timedelta(minutes=5)
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/+:-]{2,511}$")
class TaskLineageSupersessionError(RuntimeError):
def __init__(self, code: str, http_status: int = 409):
super().__init__(code)
self.code = code
self.http_status = int(http_status)
class SignatureVerifier(Protocol):
def verify(self, *, namespace: str, payload: bytes, signature: str, key_id: str) -> bool: ...
def canonical_json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
def sha256_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _fail(code: str, status: int = 409) -> None:
raise TaskLineageSupersessionError(code, status)
def _utc(value: Any, field: str) -> datetime:
if not isinstance(value, str) or not value or len(value) > 80:
_fail(field.upper() + "_INVALID", 400)
try:
result = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise TaskLineageSupersessionError(field.upper() + "_INVALID", 400) from exc
if result.tzinfo is None:
_fail(field.upper() + "_INVALID", 400)
return result.astimezone(timezone.utc)
def _text(value: Any, field: str, *, maximum: int = 1024) -> str:
if not isinstance(value, str) or not value or len(value) > maximum:
_fail(field.upper() + "_INVALID", 400)
return value
def decision_payload(raw: Mapping[str, Any]) -> dict[str, Any]:
return {key: raw[key] for key in raw if key != "signature"}
DECISION_FIELDS = {
"schema", "decision_id", "nonce", "source_task_id",
"source_task_identity_sha256", "source_metadata_sha256",
"source_expected_status", "source_expected_revision", "source_expected_attempt",
"source_expected_claimed_by", "source_expected_lease_token_hash",
"source_expected_lease_expires_at", "repository", "target_branch", "generation",
"execution_digest", "external_job_id", "child_task_id", "supersession_ordinal",
"canonical_projection_profile", "canonical_projection_version", "supersession_reason",
"owner_identity", "signature_namespace", "signature_principal", "key_id",
"issued_at", "not_before", "expires_at", "one_time_use", "signature",
}
def validate_decision(raw: Mapping[str, Any], *, now: datetime) -> dict[str, Any]:
if not isinstance(raw, Mapping) or set(raw) != DECISION_FIELDS:
_fail("DECISION_FIELDS_INVALID", 400)
exact = {
"schema": DECISION_SCHEMA,
"source_task_id": SOURCE_TASK_ID,
"source_expected_status": "EXPIRED",
"source_expected_attempt": 0,
"source_expected_claimed_by": None,
"source_expected_lease_token_hash": None,
"source_expected_lease_expires_at": None,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"execution_digest": EXECUTION_DIGEST,
"external_job_id": EXTERNAL_JOB_ID,
"child_task_id": CHILD_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
"supersession_reason": SUPERSESSION_REASON,
"owner_identity": OWNER_PRINCIPAL,
"signature_namespace": OWNER_NAMESPACE,
"signature_principal": OWNER_PRINCIPAL,
"key_id": OWNER_KEY_ID,
"one_time_use": True,
}
for field, expected in exact.items():
if raw[field] != expected:
_fail(field.upper() + "_MISMATCH", 403 if field.startswith(("owner", "signature", "key")) else 409)
for field in ("decision_id", "nonce"):
value = _text(raw[field], field, maximum=256)
if SAFE_ID.fullmatch(value) is None:
_fail(field.upper() + "_INVALID", 400)
for field in ("source_task_identity_sha256", "source_metadata_sha256"):
if not isinstance(raw[field], str) or HEX64.fullmatch(raw[field]) is None:
_fail(field.upper() + "_INVALID", 400)
if type(raw["source_expected_revision"]) is not int or raw["source_expected_revision"] < 1:
_fail("SOURCE_EXPECTED_REVISION_INVALID", 400)
_text(raw["signature"], "signature", maximum=32768)
issued_at = _utc(raw["issued_at"], "issued_at")
not_before = _utc(raw["not_before"], "not_before")
expires_at = _utc(raw["expires_at"], "expires_at")
if issued_at > expires_at or not_before > expires_at:
_fail("DECISION_TIME_ORDER_INVALID", 403)
if expires_at - issued_at > MAX_TTL or expires_at <= issued_at:
_fail("DECISION_TTL_INVALID", 403)
now = now.astimezone(timezone.utc)
if now < not_before or now >= expires_at:
_fail("DECISION_EXPIRED_OR_NOT_YET_VALID", 403)
if issued_at > now + timedelta(seconds=5) or now - issued_at > MAX_TTL:
_fail("DECISION_ISSUED_AT_INVALID", 403)
return {**dict(raw), "issued_at_dt": issued_at, "not_before_dt": not_before, "expires_at_dt": expires_at}
def _default_identity_digest(row: sqlite3.Row) -> str:
try:
from aios_expired_task_recovery import task_identity_digest
except ImportError as exc:
raise TaskLineageSupersessionError("FORMAL_TASK_IDENTITY_HELPER_UNAVAILABLE", 503) from exc
return str(task_identity_digest(row))
def _json_object(row: sqlite3.Row, field: str) -> dict[str, Any]:
try:
value = json.loads(row[field])
except (TypeError, ValueError, KeyError) as exc:
raise TaskLineageSupersessionError("SOURCE_TASK_JSON_INVALID", 409) from exc
if not isinstance(value, Mapping):
_fail("SOURCE_TASK_JSON_INVALID")
return dict(value)
def _verify_source(row: sqlite3.Row, d: Mapping[str, Any], identity_digest: Callable[[sqlite3.Row], str]) -> None:
if row["task_id"] != SOURCE_TASK_ID:
_fail("SOURCE_TASK_ID_MISMATCH")
cas = (
row["status"] == "EXPIRED"
and int(row["revision"]) == d["source_expected_revision"]
and int(row["attempt"]) == 0
and row["claimed_by"] is None
and row["lease_token_hash"] is None
and row["lease_expires_at"] is None
)
if not cas:
_fail("SOURCE_TASK_EXACT_CAS_MISMATCH")
metadata_raw = row["metadata_json"]
if not isinstance(metadata_raw, str) or sha256_text(metadata_raw) != d["source_metadata_sha256"]:
_fail("SOURCE_METADATA_DIGEST_MISMATCH")
if identity_digest(row) != d["source_task_identity_sha256"]:
_fail("SOURCE_TASK_IDENTITY_DIGEST_MISMATCH")
metadata = _json_object(row, "metadata_json")
execution = metadata.get("execution")
constraints = _json_object(row, "constraints_json")
if not isinstance(execution, Mapping):
_fail("SOURCE_EXECUTION_BINDING_INVALID")
expected = {
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"execution_digest": EXECUTION_DIGEST,
"external_job_id": EXTERNAL_JOB_ID,
"task_id": SOURCE_TASK_ID,
}
if any(execution.get(key) != value for key, value in expected.items()):
_fail("SOURCE_EXECUTION_BINDING_MISMATCH")
if constraints.get("generation") != GENERATION:
_fail("SOURCE_GENERATION_CONSTRAINT_MISMATCH")
class TaskLineageSupersessionController:
def __init__(self, db_path: str | Path, verifier: SignatureVerifier, *, clock=None, identity_digest=None):
self.db_path = Path(db_path)
self.verifier = verifier
self.clock = clock or (lambda: datetime.now(timezone.utc))
self.identity_digest = identity_digest or _default_identity_digest
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.db_path, timeout=30, isolation_level=None)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys=ON")
connection.execute("PRAGMA busy_timeout=30000")
return connection
def _now(self) -> datetime:
value = self.clock()
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
@staticmethod
def _assert_schema(connection: sqlite3.Connection) -> None:
required = {"tasks", "events", "task_lineage_supersessions", "task_lineage_supersession_receipts"}
actual = {row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")}
if not required.issubset(actual):
_fail("SUPERSESSION_SCHEMA_UNAVAILABLE", 503)
columns = {row[1] for row in connection.execute("PRAGMA table_info(tasks)")}
task_columns = {
"task_id", "goal", "instructions", "success_criteria_json",
"required_capabilities_json", "constraints_json", "metadata_json", "deadline",
"priority", "max_attempts", "attempt", "status", "claimed_by",
"lease_token_hash", "lease_expires_at", "result_json", "result_digest",
"result_idempotency_key", "verification_json", "revision", "created_at", "updated_at",
}
if not task_columns.issubset(columns):
_fail("TASK_SCHEMA_UNSUPPORTED", 503)
def supersede(self, raw: Mapping[str, Any]) -> dict[str, Any]:
now = self._now()
decision = validate_decision(raw, now=now)
payload_bytes = canonical_json(decision_payload(raw)).encode("utf-8")
if not self.verifier.verify(
namespace=OWNER_NAMESPACE, payload=payload_bytes,
signature=decision["signature"], key_id=decision["key_id"],
):
_fail("DECISION_SIGNATURE_INVALID", 403)
decision_sha = hashlib.sha256(payload_bytes).hexdigest()
now_text = now.isoformat()
with closing(self._connect()) as connection:
connection.execute("BEGIN IMMEDIATE")
try:
self._assert_schema(connection)
replay = connection.execute(
"SELECT owner_decision_sha256,receipt_json FROM task_lineage_supersession_receipts WHERE decision_id=? OR nonce=?",
(decision["decision_id"], decision["nonce"]),
).fetchone()
if replay is not None:
if replay["owner_decision_sha256"] != decision_sha:
_fail("DECISION_REPLAY_MISMATCH")
result = json.loads(replay["receipt_json"])
connection.commit()
return {**result, "idempotent": True}
source = connection.execute("SELECT * FROM tasks WHERE task_id=?", (SOURCE_TASK_ID,)).fetchone()
if source is None:
_fail("SOURCE_TASK_NOT_FOUND", 404)
_verify_source(source, decision, self.identity_digest)
if connection.execute(
"SELECT 1 FROM task_lineage_supersessions WHERE source_task_id=? AND supersession_ordinal=?",
(SOURCE_TASK_ID, SUPERSESSION_ORDINAL),
).fetchone():
_fail("SUPERSESSION_ALREADY_EXISTS")
if connection.execute("SELECT 1 FROM tasks WHERE task_id=?", (CHILD_TASK_ID,)).fetchone():
_fail("CHILD_TASK_ALREADY_EXISTS")
connection.execute(
"""INSERT INTO tasks(
task_id,goal,instructions,success_criteria_json,required_capabilities_json,
constraints_json,metadata_json,deadline,priority,max_attempts,attempt,status,
claimed_by,lease_token_hash,lease_expires_at,result_json,result_digest,
result_idempotency_key,verification_json,revision,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
CHILD_TASK_ID, source["goal"], source["instructions"], source["success_criteria_json"],
"[]", source["constraints_json"], "{}", None, source["priority"], source["max_attempts"],
0, "QUEUED", None, None, None, None, None, None, None, 1, now_text, now_text,
),
)
connection.execute(
"""INSERT INTO task_lineage_supersessions(
source_task_id,supersession_ordinal,child_task_id,source_task_identity_sha256,
source_metadata_sha256,owner_decision_sha256,decision_id,created_at)
VALUES(?,?,?,?,?,?,?,?)""",
(SOURCE_TASK_ID, SUPERSESSION_ORDINAL, CHILD_TASK_ID,
decision["source_task_identity_sha256"], decision["source_metadata_sha256"],
decision_sha, decision["decision_id"], now_text),
)
event_payload = {
"task_id": CHILD_TASK_ID, "supersedes_task_id": SOURCE_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"source_task_identity_sha256": decision["source_task_identity_sha256"],
"source_metadata_sha256": decision["source_metadata_sha256"],
"owner_supersession_decision_sha256": decision_sha,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
}
connection.execute(
"INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) VALUES(?,?,?,?,?)",
(CHILD_TASK_ID, None, "TASK_CREATED", canonical_json(event_payload), now_text),
)
receipt = {
"schema": RECEIPT_SCHEMA, "status": "SUPERSEDED",
"decision_id": decision["decision_id"], "owner_decision_sha256": decision_sha,
"source_task_id": SOURCE_TASK_ID, "source_task_identity_sha256": decision["source_task_identity_sha256"],
"source_metadata_sha256": decision["source_metadata_sha256"],
"source_status": "EXPIRED", "source_revision": decision["source_expected_revision"],
"source_attempt": 0, "source_history_preserved": True,
"child_task_id": CHILD_TASK_ID, "supersession_ordinal": SUPERSESSION_ORDINAL,
"child_status": "QUEUED", "child_revision": 1, "child_attempt": 0,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
"exact_identity_cas": True, "one_time_use": True,
"owner_signature_namespace": OWNER_NAMESPACE,
"owner_signature_principal": OWNER_PRINCIPAL,
"owner_signature_key_id": OWNER_KEY_ID,
"owner_signature_sha256": sha256_text(decision["signature"]),
}
connection.execute(
"""INSERT INTO task_lineage_supersession_receipts(
decision_id,nonce,source_task_id,supersession_ordinal,child_task_id,
owner_decision_sha256,decision_json,receipt_json,consumed_at)
VALUES(?,?,?,?,?,?,?,?,?)""",
(decision["decision_id"], decision["nonce"], SOURCE_TASK_ID, SUPERSESSION_ORDINAL,
CHILD_TASK_ID, decision_sha, canonical_json(dict(raw)), canonical_json(receipt), now_text),
)
connection.commit()
return {**receipt, "idempotent": False}
except Exception:
connection.rollback()
raise
@@ -0,0 +1,548 @@
"""Canonical initial projection for the exact Single-Host supersession child.
The source business execution digest is lineage evidence. It is deliberately
not used as the child execution-envelope digest: the canonical envelope binds
the child Task id and therefore has its own server-derived digest.
"""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
import sqlite3
from typing import Any, Mapping
from aios_agent_task_initial_execution_projection import (
InitialExecutionProjectionError,
InitialExecutionProjectionRequest,
InitialExecutionProjectionStore,
RUNTIME_CAPABILITY_RE,
_now,
_require_string,
_required_authorization_id,
)
from aios_online_agent_execution import (
ExecutionProjectionError,
GovernedExecutionBinding,
validate_execution_envelope,
)
from aios_task_lineage_supersession import (
CANONICAL_PROFILE,
CANONICAL_PROFILE_VERSION,
CHILD_TASK_ID,
DECISION_SCHEMA,
EXECUTION_DIGEST as SOURCE_BUSINESS_EXECUTION_DIGEST,
EXTERNAL_JOB_ID,
GENERATION,
OWNER_KEY_ID,
OWNER_NAMESPACE,
OWNER_PRINCIPAL,
RECEIPT_SCHEMA as SUPERSESSION_RECEIPT_SCHEMA,
REPOSITORY,
SOURCE_TASK_ID,
SUPERSESSION_ORDINAL,
TARGET_BRANCH,
canonical_json,
decision_payload,
)
PROJECTION_INTENT_SCHEMA = "aios.task-lineage-supersession-initial-projection-intent.v1"
PROJECTION_RECEIPT_SCHEMA = "aios.task-lineage-supersession-initial-projection-receipt.v1"
PROJECTION_IDEMPOTENCY_KEY = "task-lineage-supersession-s1-initial-projection-v1"
PROJECTION_REASON = "canonical-initial-projection-for-formal-supersession-child"
PROJECTION_TABLE = "task_lineage_supersession_initial_projections"
def _sha_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _sha_json(value: Any) -> str:
return _sha_text(canonical_json(value))
def _json_object(raw: Any, code: str) -> dict[str, Any]:
if not isinstance(raw, str):
raise InitialExecutionProjectionError(code, 409)
try:
value = json.loads(raw)
except (TypeError, ValueError) as exc:
raise InitialExecutionProjectionError(code, 409) from exc
if not isinstance(value, Mapping):
raise InitialExecutionProjectionError(code, 409)
return dict(value)
@dataclass(frozen=True)
class SupersessionChildProjectionIntent:
"""Caller-selectable portion of the exact child projection.
Task identity, CAS values, Job identity, projection profile, reason and
idempotency are intentionally absent from the request and server-owned.
"""
adapter: str
required_capability: str
authorization_id: str
@classmethod
def from_mapping(
cls, value: Mapping[str, Any]
) -> "SupersessionChildProjectionIntent":
if not isinstance(value, Mapping) or set(value) != {
"adapter",
"required_capability",
"authorization_id",
}:
raise InitialExecutionProjectionError(
"SUPERSESSION_INITIAL_PROJECTION_FIELDS_INVALID", 400
)
adapter = _require_string(value["adapter"], "RUNTIME_ADAPTER_INVALID")
capability = _require_string(
value["required_capability"],
"RUNTIME_CAPABILITY_INVALID",
RUNTIME_CAPABILITY_RE,
)
if adapter not in {"hermes", "openclaw"} or capability != f"agent.runtime.{adapter}":
raise InitialExecutionProjectionError("RUNTIME_CAPABILITY_MISMATCH", 400)
authorization_id = _require_string(
value["authorization_id"], "AUTHORIZATION_ID_INVALID"
)
return cls(adapter, capability, authorization_id)
def public(self) -> dict[str, Any]:
return {
"schema": PROJECTION_INTENT_SCHEMA,
"source_task_id": SOURCE_TASK_ID,
"child_task_id": CHILD_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
"expected_status": "QUEUED",
"expected_revision": 1,
"expected_attempt": 0,
"expected_metadata_sha256": _sha_json({}),
"expected_required_capabilities_sha256": _sha_json([]),
"expected_job_id": EXTERNAL_JOB_ID,
"adapter": self.adapter,
"required_capability": self.required_capability,
"authorization_id": self.authorization_id,
"idempotency_key": PROJECTION_IDEMPOTENCY_KEY,
"reason": PROJECTION_REASON,
}
def initial_request(self, canonical_execution_digest: str) -> InitialExecutionProjectionRequest:
return InitialExecutionProjectionRequest.from_mapping(
{
"expected_status": "QUEUED",
"expected_revision": 1,
"expected_attempt": 0,
"expected_metadata_sha256": _sha_json({}),
"expected_required_capabilities_sha256": _sha_json([]),
"expected_job_id": EXTERNAL_JOB_ID,
"expected_execution_digest": canonical_execution_digest,
"required_capability": self.required_capability,
"adapter": self.adapter,
"authorization_id": self.authorization_id,
"idempotency_key": PROJECTION_IDEMPOTENCY_KEY,
"reason": PROJECTION_REASON,
}
)
def _table_names(connection: sqlite3.Connection) -> set[str]:
return {
str(row[0])
for row in connection.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
)
}
def _load_lineage_context(
connection: sqlite3.Connection, *, expect_projected: bool
) -> dict[str, Any]:
required_tables = {
"tasks",
"events",
"task_lineage_supersessions",
"task_lineage_supersession_receipts",
PROJECTION_TABLE,
}
if not required_tables.issubset(_table_names(connection)):
raise InitialExecutionProjectionError(
"SUPERSESSION_INITIAL_PROJECTION_SCHEMA_UNAVAILABLE", 503
)
relation = connection.execute(
"SELECT * FROM task_lineage_supersessions "
"WHERE source_task_id=? AND supersession_ordinal=? AND child_task_id=?",
(SOURCE_TASK_ID, SUPERSESSION_ORDINAL, CHILD_TASK_ID),
).fetchone()
if relation is None:
raise InitialExecutionProjectionError("FORMAL_SUPERSESSION_REQUIRED", 409)
receipt_row = connection.execute(
"SELECT * FROM task_lineage_supersession_receipts "
"WHERE decision_id=? AND source_task_id=? AND supersession_ordinal=? "
"AND child_task_id=?",
(
relation["decision_id"],
SOURCE_TASK_ID,
SUPERSESSION_ORDINAL,
CHILD_TASK_ID,
),
).fetchone()
if receipt_row is None:
raise InitialExecutionProjectionError("SUPERSESSION_RECEIPT_REQUIRED", 409)
decision = _json_object(receipt_row["decision_json"], "SUPERSESSION_DECISION_INVALID")
receipt = _json_object(receipt_row["receipt_json"], "SUPERSESSION_RECEIPT_INVALID")
decision_sha256 = _sha_text(canonical_json(decision_payload(decision)))
exact_decision = {
"schema": DECISION_SCHEMA,
"source_task_id": SOURCE_TASK_ID,
"child_task_id": CHILD_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_id": EXTERNAL_JOB_ID,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
"owner_identity": OWNER_PRINCIPAL,
"signature_namespace": OWNER_NAMESPACE,
"signature_principal": OWNER_PRINCIPAL,
"key_id": OWNER_KEY_ID,
"one_time_use": True,
}
if any(decision.get(key) != expected for key, expected in exact_decision.items()):
raise InitialExecutionProjectionError(
"SUPERSESSION_DECISION_SCOPE_MISMATCH", 409
)
if (
decision_sha256 != relation["owner_decision_sha256"]
or decision_sha256 != receipt_row["owner_decision_sha256"]
or decision.get("source_task_identity_sha256")
!= relation["source_task_identity_sha256"]
or decision.get("source_metadata_sha256") != relation["source_metadata_sha256"]
or receipt.get("schema") != SUPERSESSION_RECEIPT_SCHEMA
or receipt.get("owner_decision_sha256") != decision_sha256
or receipt.get("source_task_id") != SOURCE_TASK_ID
or receipt.get("child_task_id") != CHILD_TASK_ID
or receipt.get("supersession_ordinal") != SUPERSESSION_ORDINAL
):
raise InitialExecutionProjectionError(
"SUPERSESSION_LINEAGE_EVIDENCE_MISMATCH", 409
)
events = connection.execute(
"SELECT event_type,payload_json FROM events WHERE task_id=? ORDER BY event_id",
(CHILD_TASK_ID,),
).fetchall()
expected_events = ["TASK_CREATED"]
if expect_projected:
expected_events.append("TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED")
if [str(row["event_type"]) for row in events] != expected_events:
raise InitialExecutionProjectionError(
"SUPERSESSION_CHILD_EVENT_CHAIN_INVALID", 409
)
created = _json_object(events[0]["payload_json"], "SUPERSESSION_CREATED_EVENT_INVALID")
exact_created = {
"task_id": CHILD_TASK_ID,
"supersedes_task_id": SOURCE_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"source_task_identity_sha256": relation["source_task_identity_sha256"],
"source_metadata_sha256": relation["source_metadata_sha256"],
"owner_supersession_decision_sha256": decision_sha256,
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
}
if any(created.get(key) != expected for key, expected in exact_created.items()):
raise InitialExecutionProjectionError(
"SUPERSESSION_CREATED_EVENT_SCOPE_MISMATCH", 409
)
return {
"source_task_identity_sha256": relation["source_task_identity_sha256"],
"source_metadata_sha256": relation["source_metadata_sha256"],
"owner_decision_sha256": decision_sha256,
}
def _validate_canonical_child_execution(
resolved: Mapping[str, Any],
*,
intent: SupersessionChildProjectionIntent,
runtime_input: Mapping[str, Any],
) -> dict[str, Any]:
try:
binding = GovernedExecutionBinding.from_mapping(resolved)
execution = validate_execution_envelope(resolved, binding=binding)
except ExecutionProjectionError as exc:
raise InitialExecutionProjectionError(
"SUPERSESSION_CANONICAL_EXECUTION_INVALID", 409
) from exc
exact_binding = {
"task_id": CHILD_TASK_ID,
"job_id": EXTERNAL_JOB_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
}
if any(execution.get(key) != expected for key, expected in exact_binding.items()):
raise InitialExecutionProjectionError(
"SUPERSESSION_CANONICAL_EXECUTION_BINDING_MISMATCH", 409
)
if (
execution.get("action") != "agent_runtime"
or execution.get("adapter") != intent.adapter
or execution.get("capability_scope") != [intent.required_capability]
or execution.get("runtime_input") != dict(runtime_input)
or execution.get("change_set") != []
or execution.get("verification") != []
or execution.get("governance", {}).get("writer_lease") is not False
):
raise InitialExecutionProjectionError(
"SUPERSESSION_AGENT_RUNTIME_ENVELOPE_INVALID", 409
)
if execution["execution_digest"] == SOURCE_BUSINESS_EXECUTION_DIGEST:
raise InitialExecutionProjectionError(
"CHILD_CANONICAL_DIGEST_COLLIDES_WITH_SOURCE_BUSINESS_DIGEST", 409
)
return execution
class _SupersessionProjectionTransactionStore(InitialExecutionProjectionStore):
"""Per-invocation store that adds immutable lineage evidence atomically."""
allows_supersession_child = True
def __init__(self, db_path, resolved: Mapping[str, Any], intent):
self._resolved = dict(resolved)
self.intent = intent
super().__init__(db_path, self._resolve_once)
def _resolve_once(self, *_args):
return dict(self._resolved)
def _projection_event_extension(
self, connection, task_id, request, execution, applied_revision
):
if task_id != CHILD_TASK_ID:
raise InitialExecutionProjectionError(
"SUPERSESSION_CHILD_TASK_ID_MISMATCH", 409
)
context = _load_lineage_context(connection, expect_projected=False)
return {
"source_task_id": SOURCE_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"source_task_identity_sha256": context["source_task_identity_sha256"],
"source_metadata_sha256": context["source_metadata_sha256"],
"owner_supersession_decision_sha256": context["owner_decision_sha256"],
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"canonical_child_execution_digest": execution["execution_digest"],
"canonical_projection_profile": CANONICAL_PROFILE,
"canonical_projection_version": CANONICAL_PROFILE_VERSION,
}
def _insert_projection_extension(
self,
connection,
task_id,
request,
execution,
event_payload,
applied_revision,
):
context = _load_lineage_context(connection, expect_projected=True)
task = connection.execute(
"SELECT metadata_json,required_capabilities_json FROM tasks WHERE task_id=?",
(CHILD_TASK_ID,),
).fetchone()
if task is None:
raise InitialExecutionProjectionError("TASK_NOT_FOUND", 404)
intent_json = canonical_json(self.intent.public())
intent_sha256 = _sha_text(intent_json)
receipt = {
"schema": PROJECTION_RECEIPT_SCHEMA,
"projection_id": "sha256:" + intent_sha256,
"task_id": CHILD_TASK_ID,
"source_task_id": SOURCE_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"source_task_identity_sha256": context["source_task_identity_sha256"],
"source_metadata_sha256": context["source_metadata_sha256"],
"owner_supersession_decision_sha256": context["owner_decision_sha256"],
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"canonical_child_execution_digest": execution["execution_digest"],
"external_job_id": EXTERNAL_JOB_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"adapter": self.intent.adapter,
"required_capability": self.intent.required_capability,
"expected_revision": 1,
"applied_revision": applied_revision,
"attempt": 0,
"metadata_sha256": _sha_text(task["metadata_json"]),
"required_capabilities_sha256": _sha_text(
task["required_capabilities_json"]
),
"projection_event_sha256": _sha_json(dict(event_payload)),
}
receipt_json = canonical_json(receipt)
receipt_sha256 = _sha_text(receipt_json)
connection.execute(
f"""INSERT INTO {PROJECTION_TABLE}(
projection_id,task_id,source_task_id,supersession_ordinal,
intent_sha256,intent_json,owner_decision_sha256,
source_task_identity_sha256,source_metadata_sha256,
source_business_execution_digest,canonical_child_execution_digest,
external_job_id,repository,target_branch,generation,adapter,
required_capability,expected_revision,applied_revision,metadata_sha256,
required_capabilities_sha256,projection_event_sha256,receipt_json,
receipt_sha256,created_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
receipt["projection_id"],
CHILD_TASK_ID,
SOURCE_TASK_ID,
SUPERSESSION_ORDINAL,
intent_sha256,
intent_json,
context["owner_decision_sha256"],
context["source_task_identity_sha256"],
context["source_metadata_sha256"],
SOURCE_BUSINESS_EXECUTION_DIGEST,
execution["execution_digest"],
EXTERNAL_JOB_ID,
REPOSITORY,
TARGET_BRANCH,
GENERATION,
self.intent.adapter,
self.intent.required_capability,
1,
applied_revision,
receipt["metadata_sha256"],
receipt["required_capabilities_sha256"],
receipt["projection_event_sha256"],
receipt_json,
receipt_sha256,
_now(),
),
)
class SupersessionChildInitialProjectionStore:
"""Server-owned adapter from exact supersession lineage to current ABI."""
def __init__(self, db_path, resolver):
self.db_path = db_path
self.resolver = resolver
def _connect(self):
connection = sqlite3.connect(self.db_path, timeout=30.0)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys=ON")
connection.execute("PRAGMA busy_timeout=30000")
return connection
@staticmethod
def _replay(connection, intent: SupersessionChildProjectionIntent):
if PROJECTION_TABLE not in _table_names(connection):
raise InitialExecutionProjectionError(
"SUPERSESSION_INITIAL_PROJECTION_SCHEMA_UNAVAILABLE", 503
)
row = connection.execute(
f"SELECT * FROM {PROJECTION_TABLE} WHERE task_id=?", (CHILD_TASK_ID,)
).fetchone()
if row is None:
tables = _table_names(connection)
if "task_initial_execution_projections" in tables:
generic = connection.execute(
"SELECT 1 FROM task_initial_execution_projections "
"WHERE task_id=? AND idempotency_key=?",
(CHILD_TASK_ID, PROJECTION_IDEMPOTENCY_KEY),
).fetchone()
if generic is not None:
raise InitialExecutionProjectionError(
"SUPERSESSION_PROJECTION_RECEIPT_MISSING", 500
)
return None
intent_sha256 = _sha_json(intent.public())
if row["intent_sha256"] != intent_sha256:
raise InitialExecutionProjectionError(
"SUPERSESSION_INITIAL_PROJECTION_INTENT_CONFLICT", 409
)
receipt = _json_object(
row["receipt_json"], "SUPERSESSION_PROJECTION_RECEIPT_INVALID"
)
if (
_sha_text(row["receipt_json"]) != row["receipt_sha256"]
or receipt.get("schema") != PROJECTION_RECEIPT_SCHEMA
or receipt.get("canonical_child_execution_digest")
!= row["canonical_child_execution_digest"]
):
raise InitialExecutionProjectionError(
"SUPERSESSION_PROJECTION_RECEIPT_DIGEST_MISMATCH", 500
)
_load_lineage_context(connection, expect_projected=True)
task = connection.execute(
"SELECT * FROM tasks WHERE task_id=?", (CHILD_TASK_ID,)
).fetchone()
if (
task is None
or task["status"] != "QUEUED"
or int(task["revision"]) != int(row["applied_revision"])
or int(task["attempt"]) != 0
or task["claimed_by"] is not None
or task["lease_token_hash"] is not None
or task["lease_expires_at"] is not None
or _sha_text(task["metadata_json"]) != row["metadata_sha256"]
or _sha_text(task["required_capabilities_json"])
!= row["required_capabilities_sha256"]
):
raise InitialExecutionProjectionError(
"SUPERSESSION_PROJECTED_TASK_CAS_MISMATCH", 409
)
return {**receipt, "idempotent": True}
def project(self, value: Mapping[str, Any]) -> dict[str, Any]:
intent = SupersessionChildProjectionIntent.from_mapping(value)
if intent.authorization_id != _required_authorization_id():
raise InitialExecutionProjectionError("AUTHORIZATION_ID_MISMATCH", 403)
with self._connect() as connection:
replay = self._replay(connection, intent)
if replay is not None:
return replay
probe_request = intent.initial_request("0" * 64)
probe_store = _SupersessionProjectionTransactionStore(
self.db_path, {}, intent
)
task, _, _, _ = probe_store._load_and_validate(
connection, CHILD_TASK_ID, probe_request
)
_load_lineage_context(connection, expect_projected=False)
runtime_input = probe_store._runtime_input(task)
resolved = dict(
self.resolver(
CHILD_TASK_ID,
EXTERNAL_JOB_ID,
intent.required_capability,
intent.adapter,
runtime_input,
)
)
execution = _validate_canonical_child_execution(
resolved, intent=intent, runtime_input=runtime_input
)
store = _SupersessionProjectionTransactionStore(
self.db_path, resolved, intent
)
result = store.project(
CHILD_TASK_ID, intent.initial_request(execution["execution_digest"])
)
with self._connect() as connection:
receipt = self._replay(connection, intent)
if receipt is None:
raise InitialExecutionProjectionError(
"SUPERSESSION_PROJECTION_RECEIPT_MISSING", 500
)
return {**receipt, "idempotent": bool(result["idempotent"])}
@@ -0,0 +1,221 @@
"""Fixed Owner operation for one exact Task-lineage supersession.
This staging overlay deliberately extends the installed single-host operational
authority instead of exposing a generic signing primitive. The caller can
only provide the three live CAS values which cannot be known at build time.
Every authority, lineage and execution field is constructed server-side.
"""
from __future__ import annotations
import re
import secrets
from typing import Any, Callable, Mapping
from aios_single_host_authority import OWNER_IDENTITY, OWNER_NAMESPACE, sha256_json
from aios_single_host_operational_authority import (
OperationalAuthorityError,
OperationalOwnerAuthority,
OperationalPolicy,
SignedOperationalArtifact,
)
TASK_LINEAGE_SUPERSESSION_OPERATION = "issue_task_lineage_supersession"
TASK_LINEAGE_SUPERSESSION_SCHEMA = "aios.task-lineage-supersession-decision.v1"
SOURCE_TASK_ID = "TASK-SINGLEHOST-TRUSTROOT-P0-20260904"
CHILD_TASK_ID = SOURCE_TASK_ID + "-S1"
SUPERSESSION_ORDINAL = 1
REPOSITORY = "aios/aios"
TARGET_BRANCH = "agent/single-host-trust-root-p0-20260904"
GENERATION = 1
SOURCE_BUSINESS_EXECUTION_DIGEST = (
"e99bda6a251832eb3865271eb5687af7eea44dd261144cdcdfa57104c0780536"
)
EXTERNAL_JOB_IDENTITY = "job-4e10f92a6a5742518a9f1d8745c845ab"
CANONICAL_PROJECTION_PROFILE = "aios.task-initial-execution.v1"
CANONICAL_PROJECTION_VERSION = 1
SUPERSESSION_REASON = "formal-lineage-supersession-after-expired-unclaimed-source"
DECISION_TTL_SECONDS = 300
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SAFE_TOKEN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/+:-]{2,511}$")
def _fail(code: str) -> None:
raise OperationalAuthorityError(code)
def _assert_fixed_policy(policy: OperationalPolicy) -> None:
"""Refuse issuance if the live policy is not the approved source scope."""
policy.validate()
exact = {
"task_id": SOURCE_TASK_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"execution_plan_sha256": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_identity": EXTERNAL_JOB_IDENTITY,
}
if any(getattr(policy, field) != expected for field, expected in exact.items()):
_fail("TASK_LINEAGE_SUPERSESSION_POLICY_MISMATCH")
if policy.maximum_recovery_decision_ttl_seconds < DECISION_TTL_SECONDS:
_fail("TASK_LINEAGE_SUPERSESSION_POLICY_TTL_TOO_NARROW")
TASK_LINEAGE_SUPERSESSION_FIELDS = {
"schema",
"decision_id",
"nonce",
"source_task_id",
"source_task_identity_sha256",
"source_metadata_sha256",
"source_expected_status",
"source_expected_revision",
"source_expected_attempt",
"source_expected_claimed_by",
"source_expected_lease_token_hash",
"source_expected_lease_expires_at",
"repository",
"target_branch",
"generation",
"source_business_execution_digest",
"external_job_identity",
"child_task_id",
"supersession_ordinal",
"canonical_projection_profile",
"canonical_projection_version",
"supersession_reason",
"owner_identity",
"policy_digest",
"one_time_use",
"issued_at",
"not_before",
"expires_at",
"signature_namespace",
"signature_key_id",
"canonical_payload_digest",
}
def verify_task_lineage_supersession_scope(
decision: Mapping[str, Any],
policy: OperationalPolicy,
*,
now: int,
) -> None:
"""Validate a signature-verified decision against the fixed source scope."""
_assert_fixed_policy(policy)
if not isinstance(decision, Mapping) or set(decision) != TASK_LINEAGE_SUPERSESSION_FIELDS:
_fail("TASK_LINEAGE_SUPERSESSION_FIELDS_INVALID")
exact = {
"schema": TASK_LINEAGE_SUPERSESSION_SCHEMA,
"source_task_id": SOURCE_TASK_ID,
"source_expected_status": "EXPIRED",
"source_expected_attempt": 0,
"source_expected_claimed_by": None,
"source_expected_lease_token_hash": None,
"source_expected_lease_expires_at": None,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_identity": EXTERNAL_JOB_IDENTITY,
"child_task_id": CHILD_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"canonical_projection_profile": CANONICAL_PROJECTION_PROFILE,
"canonical_projection_version": CANONICAL_PROJECTION_VERSION,
"supersession_reason": SUPERSESSION_REASON,
"owner_identity": OWNER_IDENTITY,
"policy_digest": policy.digest,
"one_time_use": True,
"signature_namespace": OWNER_NAMESPACE,
}
if any(decision.get(field) != expected for field, expected in exact.items()):
_fail("TASK_LINEAGE_SUPERSESSION_SCOPE_INVALID")
if decision.get("source_task_identity_sha256") != policy.task_identity_sha256:
_fail("TASK_LINEAGE_SUPERSESSION_SOURCE_IDENTITY_INVALID")
for field in ("source_task_identity_sha256", "source_metadata_sha256"):
if not isinstance(decision.get(field), str) or HEX64.fullmatch(decision[field]) is None:
_fail("TASK_LINEAGE_SUPERSESSION_DIGEST_INVALID")
if type(decision.get("source_expected_revision")) is not int or not 1 <= decision["source_expected_revision"] < 2**63:
_fail("TASK_LINEAGE_SUPERSESSION_REVISION_INVALID")
for field in ("decision_id", "nonce"):
if not isinstance(decision.get(field), str) or SAFE_TOKEN.fullmatch(decision[field]) is None:
_fail("TASK_LINEAGE_SUPERSESSION_GENERATED_ID_INVALID")
for field in ("issued_at", "not_before", "expires_at"):
if type(decision.get(field)) is not int:
_fail("TASK_LINEAGE_SUPERSESSION_TIME_INVALID")
if not decision["not_before"] <= now < decision["expires_at"]:
_fail("TASK_LINEAGE_SUPERSESSION_TIME_INVALID")
if decision["issued_at"] > now + 5 or now - decision["issued_at"] > DECISION_TTL_SECONDS:
_fail("TASK_LINEAGE_SUPERSESSION_TIME_INVALID")
if decision["expires_at"] - decision["issued_at"] != DECISION_TTL_SECONDS:
_fail("TASK_LINEAGE_SUPERSESSION_TIME_INVALID")
digest_value = dict(decision)
supplied_digest = digest_value.pop("canonical_payload_digest", None)
if supplied_digest != sha256_json(digest_value):
_fail("TASK_LINEAGE_SUPERSESSION_CANONICAL_DIGEST_INVALID")
class TaskLineageSupersessionOwnerAuthority(OperationalOwnerAuthority):
"""Owner signer with one bounded constructor and no caller-defined message."""
def __init__(
self,
*args: Any,
token_hex: Callable[[int], str] = secrets.token_hex,
**kwargs: Any,
) -> None:
super().__init__(*args, **kwargs)
self._token_hex = token_hex
def issue_task_lineage_supersession(
self,
*,
source_expected_revision: int,
source_task_identity_sha256: str,
source_metadata_sha256: str,
) -> SignedOperationalArtifact:
_assert_fixed_policy(self.policy)
if type(source_expected_revision) is not int or not 1 <= source_expected_revision < 2**63:
_fail("TASK_LINEAGE_SUPERSESSION_REVISION_INVALID")
if source_task_identity_sha256 != self.policy.task_identity_sha256:
_fail("TASK_LINEAGE_SUPERSESSION_SOURCE_IDENTITY_INVALID")
if not isinstance(source_metadata_sha256, str) or HEX64.fullmatch(source_metadata_sha256) is None:
_fail("TASK_LINEAGE_SUPERSESSION_METADATA_DIGEST_INVALID")
decision_id = "task-lineage-supersession-" + self._token_hex(16)
nonce = "task-lineage-supersession-nonce-" + self._token_hex(16)
if SAFE_TOKEN.fullmatch(decision_id) is None or SAFE_TOKEN.fullmatch(nonce) is None:
_fail("TASK_LINEAGE_SUPERSESSION_GENERATED_ID_INVALID")
now = int(self.clock())
payload = {
"schema": TASK_LINEAGE_SUPERSESSION_SCHEMA,
"decision_id": decision_id,
"nonce": nonce,
"source_task_id": SOURCE_TASK_ID,
"source_task_identity_sha256": source_task_identity_sha256,
"source_metadata_sha256": source_metadata_sha256,
"source_expected_status": "EXPIRED",
"source_expected_revision": source_expected_revision,
"source_expected_attempt": 0,
"source_expected_claimed_by": None,
"source_expected_lease_token_hash": None,
"source_expected_lease_expires_at": None,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"generation": GENERATION,
"source_business_execution_digest": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_identity": EXTERNAL_JOB_IDENTITY,
"child_task_id": CHILD_TASK_ID,
"supersession_ordinal": SUPERSESSION_ORDINAL,
"canonical_projection_profile": CANONICAL_PROJECTION_PROFILE,
"canonical_projection_version": CANONICAL_PROJECTION_VERSION,
"supersession_reason": SUPERSESSION_REASON,
"owner_identity": OWNER_IDENTITY,
"policy_digest": self.policy.digest,
"one_time_use": True,
"issued_at": now,
"not_before": max(0, now - 5),
"expires_at": now + DECISION_TTL_SECONDS,
}
return self._sign(payload)
@@ -0,0 +1,190 @@
"""Staging-only server overlay for the fixed lineage Owner operation."""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
import sqlite3
import time
from typing import Any, Callable, Mapping
from aios_single_host_authority import DEFAULT_REGISTRY_PATH
from aios_single_host_operational_authority import (
OperationalMachineAuthority,
OperationalVerifier,
)
from aios_single_host_operational_authority_server import (
DEFAULT_LEDGER,
DEFAULT_MACHINE_KEY,
DEFAULT_OWNER_KEY,
DEFAULT_POLICY,
DurableIssuanceLedger,
OperationalAuthorityService as BaseOperationalAuthorityService,
_read_policy,
fail,
)
from aios_task_lineage_supersession_owner_authority import (
TASK_LINEAGE_SUPERSESSION_OPERATION,
TaskLineageSupersessionOwnerAuthority,
)
REQUEST_FIELDS = {
"source_expected_revision",
"source_task_identity_sha256",
"source_metadata_sha256",
}
class TaskLineageSupersessionIssuanceLedger(DurableIssuanceLedger):
"""Operation-scoped one-time fence layered on the existing durable ledger."""
def issue_once(
self,
*,
role: str,
operation: str,
request: Mapping[str, Any],
issuer: Callable[[], dict[str, Any]],
) -> dict[str, Any]:
request_sha = hashlib.sha256(
json.dumps(
request,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
).hexdigest()
connection = sqlite3.connect(self.path, timeout=5, isolation_level=None)
try:
connection.execute("BEGIN IMMEDIATE")
existing = connection.execute(
"SELECT 1 FROM issuance WHERE role=? AND operation=? LIMIT 1",
(role, operation),
).fetchone()
if existing is not None:
connection.rollback()
fail("TASK_LINEAGE_SUPERSESSION_ALREADY_ISSUED")
result = issuer()
artifact = result.get("artifact")
nonce = artifact.get("payload", {}).get("nonce") if isinstance(artifact, Mapping) else None
if not isinstance(nonce, str) or not nonce:
connection.rollback()
fail("TASK_LINEAGE_SUPERSESSION_ISSUER_RESULT_INVALID")
encoded = json.dumps(
result,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
connection.execute(
"INSERT INTO issuance(role,operation,nonce,request_sha256,response_json,created_at) "
"VALUES(?,?,?,?,?,?)",
(role, operation, nonce, request_sha, encoded, int(time.time())),
)
connection.commit()
return result
except sqlite3.IntegrityError:
connection.rollback()
fail("ISSUANCE_LEDGER_RACE")
except Exception:
connection.rollback()
raise
finally:
connection.close()
class OperationalAuthorityService(BaseOperationalAuthorityService):
"""Preserve all existing operations and add one exact Owner-only operation."""
def __init__(self, *args: Any, token_hex=None, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.ledger = TaskLineageSupersessionIssuanceLedger(self.ledger.path)
self._token_hex = token_hex
def _components(self):
policy = _read_policy(
self.policy_path,
require_root_owner=self.require_root_owner,
)
verifier = OperationalVerifier(
self.registry_path,
require_root_owner=self.require_root_owner,
)
if self.role == "owner":
kwargs = {}
if self._token_hex is not None:
kwargs["token_hex"] = self._token_hex
signer = TaskLineageSupersessionOwnerAuthority(
self.owner_key_path,
policy,
require_root_owner=self.require_root_owner,
clock=self.clock,
**kwargs,
)
else:
signer = OperationalMachineAuthority(
self.machine_key_path,
policy,
require_root_owner=self.require_root_owner,
clock=self.clock,
)
return policy, verifier, signer
def handle(self, request: Mapping[str, Any], *, peer_uid: int) -> dict[str, Any]:
if peer_uid not in self.allowed_uids:
fail("PEER_UID_FORBIDDEN")
if not isinstance(request, Mapping) or set(request) != {"operation", "payload"}:
fail("REQUEST_INVALID")
if request.get("operation") != TASK_LINEAGE_SUPERSESSION_OPERATION:
return super().handle(request, peer_uid=peer_uid)
if self.role != "owner":
fail("MACHINE_OPERATION_FORBIDDEN")
payload = request.get("payload")
if not isinstance(payload, Mapping) or set(payload) != REQUEST_FIELDS:
fail("REQUEST_FIELDS_INVALID")
_policy, _verifier, signer = self._components()
def issue() -> dict[str, Any]:
artifact = signer.issue_task_lineage_supersession(
source_expected_revision=payload["source_expected_revision"],
source_task_identity_sha256=payload["source_task_identity_sha256"],
source_metadata_sha256=payload["source_metadata_sha256"],
)
return {"status": "PASS", "artifact": artifact.to_dict()}
return self.ledger.issue_once(
role=self.role,
operation=TASK_LINEAGE_SUPERSESSION_OPERATION,
request=payload,
issuer=issue,
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--role", choices=("owner", "machine"), required=True)
parser.add_argument("--policy", default=str(DEFAULT_POLICY))
parser.add_argument("--registry", default=str(DEFAULT_REGISTRY_PATH))
parser.add_argument("--owner-key", default=str(DEFAULT_OWNER_KEY))
parser.add_argument("--machine-key", default=str(DEFAULT_MACHINE_KEY))
parser.add_argument("--socket")
parser.add_argument("--ledger", default=str(DEFAULT_LEDGER))
arguments = parser.parse_args()
service = OperationalAuthorityService(
role=arguments.role,
policy_path=Path(arguments.policy),
registry_path=Path(arguments.registry),
owner_key_path=Path(arguments.owner_key),
machine_key_path=Path(arguments.machine_key),
socket_path=Path(arguments.socket) if arguments.socket else None,
ledger_path=Path(arguments.ledger),
)
service.serve_forever()
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,92 @@
PRAGMA foreign_keys=ON;
CREATE TABLE IF NOT EXISTS task_lineage_supersessions (
source_task_id TEXT NOT NULL,
supersession_ordinal INTEGER NOT NULL CHECK (supersession_ordinal = 1),
child_task_id TEXT NOT NULL UNIQUE,
source_task_identity_sha256 TEXT NOT NULL,
source_metadata_sha256 TEXT NOT NULL,
owner_decision_sha256 TEXT NOT NULL UNIQUE,
decision_id TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
PRIMARY KEY (source_task_id, supersession_ordinal)
);
CREATE TABLE IF NOT EXISTS task_lineage_supersession_receipts (
decision_id TEXT PRIMARY KEY,
nonce TEXT NOT NULL UNIQUE,
source_task_id TEXT NOT NULL,
supersession_ordinal INTEGER NOT NULL,
child_task_id TEXT NOT NULL UNIQUE,
owner_decision_sha256 TEXT NOT NULL UNIQUE,
decision_json TEXT NOT NULL,
receipt_json TEXT NOT NULL,
consumed_at TEXT NOT NULL,
FOREIGN KEY (source_task_id, supersession_ordinal)
REFERENCES task_lineage_supersessions(source_task_id, supersession_ordinal)
);
CREATE TABLE IF NOT EXISTS task_claim_guard_bindings_v2 (
binding_id TEXT PRIMARY KEY,
task_id TEXT NOT NULL UNIQUE,
source_task_id TEXT NOT NULL,
supersession_ordinal INTEGER NOT NULL,
expected_revision INTEGER NOT NULL CHECK (expected_revision >= 1),
metadata_sha256 TEXT NOT NULL,
required_capabilities_sha256 TEXT NOT NULL,
worker_id TEXT NOT NULL,
execution_digest TEXT NOT NULL,
repository TEXT NOT NULL,
target_branch TEXT NOT NULL,
generation INTEGER NOT NULL,
policy_json TEXT NOT NULL,
policy_sha256 TEXT NOT NULL,
binding_json TEXT NOT NULL,
binding_sha256 TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
FOREIGN KEY (source_task_id, supersession_ordinal)
REFERENCES task_lineage_supersessions(source_task_id, supersession_ordinal)
);
CREATE TABLE IF NOT EXISTS task_claim_guard_binding_receipts_v2 (
receipt_id TEXT PRIMARY KEY,
binding_id TEXT NOT NULL UNIQUE,
task_id TEXT NOT NULL UNIQUE,
receipt_json TEXT NOT NULL,
receipt_sha256 TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
FOREIGN KEY (binding_id) REFERENCES task_claim_guard_bindings_v2(binding_id)
);
CREATE TRIGGER IF NOT EXISTS task_lineage_supersessions_no_update
BEFORE UPDATE ON task_lineage_supersessions BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_lineage_supersessions_no_delete
BEFORE DELETE ON task_lineage_supersessions BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_lineage_supersession_receipts_no_update
BEFORE UPDATE ON task_lineage_supersession_receipts BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_RECEIPT_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_lineage_supersession_receipts_no_delete
BEFORE DELETE ON task_lineage_supersession_receipts BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_RECEIPT_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_claim_guard_bindings_v2_no_update
BEFORE UPDATE ON task_claim_guard_bindings_v2 BEGIN
SELECT RAISE(ABORT, 'TASK_CLAIM_GUARD_BINDING_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_claim_guard_bindings_v2_no_delete
BEFORE DELETE ON task_claim_guard_bindings_v2 BEGIN
SELECT RAISE(ABORT, 'TASK_CLAIM_GUARD_BINDING_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_claim_guard_binding_receipts_v2_no_update
BEFORE UPDATE ON task_claim_guard_binding_receipts_v2 BEGIN
SELECT RAISE(ABORT, 'TASK_CLAIM_GUARD_BINDING_RECEIPT_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_claim_guard_binding_receipts_v2_no_delete
BEFORE DELETE ON task_claim_guard_binding_receipts_v2 BEGIN
SELECT RAISE(ABORT, 'TASK_CLAIM_GUARD_BINDING_RECEIPT_APPEND_ONLY');
END;
@@ -0,0 +1,41 @@
PRAGMA foreign_keys=ON;
CREATE TABLE IF NOT EXISTS task_lineage_supersession_initial_projections (
projection_id TEXT PRIMARY KEY,
task_id TEXT NOT NULL UNIQUE,
source_task_id TEXT NOT NULL,
supersession_ordinal INTEGER NOT NULL CHECK (supersession_ordinal = 1),
intent_sha256 TEXT NOT NULL UNIQUE,
intent_json TEXT NOT NULL,
owner_decision_sha256 TEXT NOT NULL UNIQUE,
source_task_identity_sha256 TEXT NOT NULL,
source_metadata_sha256 TEXT NOT NULL,
source_business_execution_digest TEXT NOT NULL,
canonical_child_execution_digest TEXT NOT NULL UNIQUE,
external_job_id TEXT NOT NULL,
repository TEXT NOT NULL,
target_branch TEXT NOT NULL,
generation INTEGER NOT NULL,
adapter TEXT NOT NULL,
required_capability TEXT NOT NULL,
expected_revision INTEGER NOT NULL CHECK (expected_revision = 1),
applied_revision INTEGER NOT NULL CHECK (applied_revision = 2),
metadata_sha256 TEXT NOT NULL,
required_capabilities_sha256 TEXT NOT NULL,
projection_event_sha256 TEXT NOT NULL,
receipt_json TEXT NOT NULL,
receipt_sha256 TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
FOREIGN KEY (source_task_id, supersession_ordinal)
REFERENCES task_lineage_supersessions(source_task_id, supersession_ordinal)
);
CREATE TRIGGER IF NOT EXISTS task_lineage_supersession_initial_projections_no_update
BEFORE UPDATE ON task_lineage_supersession_initial_projections BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_INITIAL_PROJECTION_APPEND_ONLY');
END;
CREATE TRIGGER IF NOT EXISTS task_lineage_supersession_initial_projections_no_delete
BEFORE DELETE ON task_lineage_supersession_initial_projections BEGIN
SELECT RAISE(ABORT, 'TASK_LINEAGE_SUPERSESSION_INITIAL_PROJECTION_APPEND_ONLY');
END;
@@ -0,0 +1,79 @@
# ADR: Bounded Immutable Repository Discovery
- Status: Implemented locally; pending governed publication and live runtime
deployment
- Date: 2026-09-02
- Scope: AIOS Repository Discovery P0
## Context
Repository discovery was needed to inspect unknown paths and compare exact
repository snapshots, while the ordinary SSH route was not an authenticated
execution plane. The discovery surface must not turn a readable runtime into a
general-purpose shell, grep, checkout, or Git writer. It also must remain safe
when a branch or tag moves between requests and when a diff is large.
## Decision
Implement a repository-scoped, read-only module behind the formal Bridge and
Actions read routes:
1. Verify the exact `aios/aios` identity before every request.
2. Resolve `HEAD`, branch, tag, or full commit input to one full commit OID;
reject ambiguous short names and revision expressions.
3. Invoke only fixed read-only Git subcommands (`rev-parse`, `ls-tree`,
`cat-file`, and `diff`) with `shell=False`, disabled prompts, and bounded
output. Stream NUL-delimited tree and diff records rather than collecting a
complete tree or patch.
4. Enforce normalized relative paths, repository allowlisting, depth/page/
match/byte limits, binary-source exclusion, and no patch response.
5. Sign pagination cursors with an HMAC over the operation, repository,
resolved OID, complete normalized query, and offset. A cursor for another
snapshot or query is invalid.
6. Return resolved commit identity for every read and blob identity for file
reads. Keep the four operations read-only and non-consequential at the
Bridge contract boundary.
7. Keep the mutating implementation workflow separate: Task and TaskPrincipal
admission, claim/heartbeat, WriterLease, CAS, and `commitFilesGoverned` are
still mandatory for source changes.
The canonical runtime paths are `/v2/repository/*`; the Actions adapter maps
the same contract to `/api/repository/*`. The OpenAPI document records the
mapping without copying a Path Item or duplicating operation IDs.
## Alternatives rejected
- Retrying arbitrary SSH credentials or inventing a new network profile would
bypass the supported execution-plane boundary.
- User-controlled `grep`, shell commands, checkout paths, or direct bare-repo
reads would make the repository and host boundary non-deterministic.
- Reusing the write route would introduce unnecessary Task/lease/credential
authority into a read-only request and make a read endpoint a write probe.
- Returning a full patch or collecting all Git output before pagination would
violate the bounded-output requirement for large repositories.
## Consequences
The API callers must carry the returned OID/cursor and must not assume that a
floating ref remains unchanged. Search may return fewer results when its byte
bound is exhausted, and a diff is an index rather than a patch. Cursor secret
configuration becomes a required runtime dependency for paginated operations.
The isolated implementation is testable against a disposable bare repository.
Windows can compile and run the portable discovery tests, but importing the
full Bridge/Actions runtime requires POSIX `fcntl`; Linux HTTP and locking
integration remains a separate verification item.
At the time of this decision, live HTTPS authority/runtime readback was
available, but the live OpenAPI/capability surface did not advertise these four
discovery operations and no authorized P0 Task envelope was present in the
execution context. Therefore this ADR does not claim a governed commit,
remote publication, or deployed runtime.
## Verification gate
This decision is complete for local implementation only after Python compile,
discovery unit/large-diff tests, and Bridge/OpenAPI/Actions drift tests pass.
It becomes live only after a formal Task, TaskPrincipal, lease, governed
commit, and independent authority/remote/API OID readback prove the exact same
commit, with all frozen write and staging invariants unchanged.
@@ -0,0 +1,38 @@
# ADR: Owner Scoped Task Lineage Supersession P0
## 决策
`TASK-SINGLEHOST-LINEAGE-SUPERSESSION-P0-20260905` 新增一个只服务于
`TASK-SINGLEHOST-TRUSTROOT-P0-20260904` 到
`TASK-SINGLEHOST-TRUSTROOT-P0-20260904-S1` 的控制面 capability。
它创建 append-only child lineage;绝不重写 source Task、metadata、revision、attempt、claim 或 lease。
## 安全边界
请求只携带 Owner decision 和 source 身份断言。控制面复用正式 Task identity helper
与现有 Owner verifier,严格校验 namespace、principal、当前 Owner key、短 TTL、nonce、
decision ID、exact-CAS 字段以及唯一 child ID/ordinal。未知字段、任意 child payload、通配
Task/repository/branch、旧 signer 或 replay 均 fail-closed。
child payload 由 canonical initial-execution profile 在服务端构造:`QUEUED`、revision 1、
attempt 0、无 claim/lease、空初始 capabilities 和 pristine metadata。child 普通 event 链
只允许 `TASK_CREATED`;lineage/decision/receipt 审计必须写入独立 append-only 表。
## 一致性和重放
单一 `BEGIN IMMEDIATE` 事务重读 source exact-CAS、检查 source+ordinal/child/decision/nonce
唯一性、插入 child、lineage、decision 和 receipt,并整体提交。任一步失败必须回滚,不能
留下半创建 child。只有 decision digest 完全相同的请求可返回同一 receipt;相同唯一键而
digest 不同必须拒绝。append-only 表禁止 UPDATE 和 DELETE。
## ClaimGuard
source 的 frozen ClaimGuard artifact 永不修改。child 的 Task identity 和 projected metadata
identity 必然不同,必须使用新建的版本化、append-only child binding/receipt,精确绑定
child task、projected revision/metadata/capabilities、worker、execution digest、repository、
branch、generation 和 external job。dispatch 只识别 source 与 exact child;缺失或重复 binding
一律拒绝。不得使用 latest、通配符或 caller 指定路径。
## 验收
live invocation 前必须在 isolated SQLite 完成 source exact state 到 canonical projection、
+171
View File
@@ -240,11 +240,177 @@ components:
additionalProperties: false
properties:
candidate_id: {type: string}
RepositoryTreeEntry:
type: object
required: [path, type, object_oid]
properties:
path: {type: string, minLength: 1, maxLength: 512}
type: {type: string, enum: [blob, tree, commit]}
object_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
mode: {type: string}
RepositoryTreeResponse:
type: object
required: [repository, resolved_oid, entries, truncated, next_cursor]
properties:
repository: {type: string, const: aios/aios}
resolved_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
entries: {type: array, items: {$ref: '#/components/schemas/RepositoryTreeEntry'}}
truncated: {type: boolean}
next_cursor: {type: string, nullable: true, maxLength: 4096}
RepositorySearchMatch:
type: object
required: [path, line_start, line_end, blob_oid, snippet]
properties:
path: {type: string, minLength: 1, maxLength: 512}
line_start: {type: integer, minimum: 1}
line_end: {type: integer, minimum: 1}
blob_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
snippet: {type: string, maxLength: 512}
RepositorySearchResponse:
type: object
required: [repository, resolved_oid, matches, scanned_bytes, max_bytes, truncated, next_cursor]
properties:
repository: {type: string, const: aios/aios}
resolved_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
matches: {type: array, items: {$ref: '#/components/schemas/RepositorySearchMatch'}}
scanned_bytes: {type: integer, minimum: 0}
max_bytes: {type: integer, minimum: 1, maximum: 1048576}
binary_blobs_skipped: {type: integer, minimum: 0}
truncated: {type: boolean}
next_cursor: {type: string, nullable: true, maxLength: 4096}
RepositoryDiffEntry:
type: object
required: [path, status, additions, deletions, old_path]
properties:
path: {type: string, minLength: 1, maxLength: 512}
status: {type: string, pattern: '^[ACDMRTUX][0-9]*$'}
additions: {type: integer, nullable: true, minimum: 0}
deletions: {type: integer, nullable: true, minimum: 0}
old_path: {type: string, nullable: true, maxLength: 512}
RepositoryDiffIndexResponse:
type: object
required: [repository, resolved_from_oid, resolved_to_oid, entries, truncated, next_cursor]
properties:
repository: {type: string, const: aios/aios}
resolved_from_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
resolved_to_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
entries: {type: array, items: {$ref: '#/components/schemas/RepositoryDiffEntry'}}
truncated: {type: boolean}
next_cursor: {type: string, nullable: true, maxLength: 4096}
RepositoryFileResponse:
type: object
required: [repository, resolved_oid, blob_oid, path, content, content_base64, content_bytes, truncated]
properties:
repository: {type: string, const: aios/aios}
resolved_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
blob_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
path: {type: string, minLength: 1, maxLength: 512}
content: {type: string}
content_base64: {type: string, format: byte}
content_bytes: {type: integer, minimum: 0}
truncated: {type: boolean}
paths:
/v2/repository/tree:
get:
operationId: listRepositoryTree
summary: List one bounded page of an immutable repository tree
security: []
x-openai-isConsequential: false
x-aios-read-only: true
x-aios-bounded: true
x-aios-repository-scoped: true
x-aios-no-shell: true
x-aios-legacy-path: /api/repository/tree
parameters:
- {in: query, name: repository, required: true, schema: {type: string, const: aios/aios}}
- {in: query, name: ref, required: true, schema: {type: string, minLength: 1, maxLength: 240}}
- {in: query, name: prefix, schema: {type: string, maxLength: 512}}
- {in: query, name: depth, schema: {type: integer, minimum: 1, maximum: 32, default: 8}}
- {in: query, name: limit, required: true, schema: {type: integer, minimum: 1, maximum: 1000}}
- {in: query, name: cursor, schema: {type: string, maxLength: 4096}}
responses:
'200': {description: Immutable tree page, content: {application/json: {schema: {$ref: '#/components/schemas/RepositoryTreeResponse'}}}}
'400': {description: Invalid repository, ref, path, bounds, or cursor}
'403': {description: Repository identity is not allowed}
'404': {description: Repository ref or path not found}
'503': {description: Authoritative repository or discovery configuration unavailable}
/v2/repository/search:
get:
operationId: searchRepositorySource
summary: Search bounded source blobs at one immutable repository revision
security: []
x-openai-isConsequential: false
x-aios-read-only: true
x-aios-bounded: true
x-aios-repository-scoped: true
x-aios-no-shell: true
x-aios-legacy-path: /api/repository/search
parameters:
- {in: query, name: repository, required: true, schema: {type: string, const: aios/aios}}
- {in: query, name: ref, required: true, schema: {type: string, minLength: 1, maxLength: 240}}
- {in: query, name: query, required: true, schema: {type: string, minLength: 1, maxLength: 512}}
- {in: query, name: path_globs, style: form, explode: true, schema: {type: array, maxItems: 16, items: {type: string, maxLength: 512}}}
- {in: query, name: max_matches, required: true, schema: {type: integer, minimum: 1, maximum: 1000}}
- {in: query, name: max_bytes, required: true, schema: {type: integer, minimum: 1, maximum: 1048576}}
- {in: query, name: cursor, schema: {type: string, maxLength: 4096}}
responses:
'200': {description: Bounded source matches with blob identity, content: {application/json: {schema: {$ref: '#/components/schemas/RepositorySearchResponse'}}}}
'400': {description: Invalid repository, ref, query, path glob, bounds, or cursor}
'403': {description: Repository identity is not allowed}
'404': {description: Repository ref not found}
'503': {description: Authoritative repository or discovery configuration unavailable}
/v2/repository/diff-index:
get:
operationId: getDiffIndex
summary: List a bounded immutable diff index without returning a patch
security: []
x-openai-isConsequential: false
x-aios-read-only: true
x-aios-bounded: true
x-aios-repository-scoped: true
x-aios-no-shell: true
x-aios-legacy-path: /api/repository/diff-index
parameters:
- {in: query, name: repository, required: true, schema: {type: string, const: aios/aios}}
- {in: query, name: from, required: true, schema: {type: string, minLength: 1, maxLength: 240}}
- {in: query, name: to, required: true, schema: {type: string, minLength: 1, maxLength: 240}}
- {in: query, name: path_prefix, schema: {type: string, maxLength: 512}}
- {in: query, name: limit, required: true, schema: {type: integer, minimum: 1, maximum: 1000}}
- {in: query, name: cursor, schema: {type: string, maxLength: 4096}}
responses:
'200': {description: Bounded diff index with immutable endpoints, content: {application/json: {schema: {$ref: '#/components/schemas/RepositoryDiffIndexResponse'}}}}
'400': {description: Invalid repository, revision, path prefix, bounds, or cursor}
'403': {description: Repository identity is not allowed}
'404': {description: Repository revision not found}
'503': {description: Authoritative repository or discovery configuration unavailable}
/v2/repository/file:
get:
operationId: readRepositoryFile
summary: Read one bounded file from an immutable repository revision
security: []
x-openai-isConsequential: false
x-aios-read-only: true
x-aios-bounded: true
x-aios-repository-scoped: true
x-aios-no-shell: true
x-aios-legacy-path: /api/repository/file
parameters:
- {in: query, name: repository, required: true, schema: {type: string, const: aios/aios}}
- {in: query, name: ref, required: true, schema: {type: string, minLength: 1, maxLength: 240}}
- {in: query, name: path, required: true, schema: {type: string, minLength: 1, maxLength: 512}}
- {in: query, name: max_bytes, schema: {type: integer, minimum: 1, maximum: 1048576, default: 262144}}
responses:
'200': {description: Bounded file content with resolved commit and blob identity, content: {application/json: {schema: {$ref: '#/components/schemas/RepositoryFileResponse'}}}}
'400': {description: Invalid repository, ref, path, or bounds}
'403': {description: Repository identity is not allowed}
'404': {description: Repository ref or file not found}
'503': {description: Authoritative repository or discovery configuration unavailable}
/v2/commit-files:
post:
operationId: commitFilesGoverned
summary: Submit one atomic governed commit containing multiple files
security: [{taskPrincipalAuth: []}]
x-aios-task-principal: Bridge verifies a fresh aios.task-principal-assertion.v1 and exact task/job/repository/branch/generation/revision/attempt/execution-digest/capability binding; caller JSON cannot provide a principal and missing verification is denied.
requestBody:
required: true
content:
@@ -254,6 +420,7 @@ paths:
additionalProperties: false
required:
- task_id
- job_id
- repository
- expected_branch_oid
- target_branch
@@ -283,8 +450,12 @@ paths:
message: {type: string, minLength: 1, maxLength: 200}
responses:
'200': {description: Governed commit job and verified branch result}
'202': {description: Existing external Job alias requires authoritative reconciliation; no replacement Job is created}
'401': {description: No verified Task principal is available}
'403': {description: Verified Task principal lacks exact task, external Job, or capability scope}
'400': {description: Contract or policy rejection}
'409': {description: Branch CAS or idempotency conflict}
'423': {description: Narrow P0 Task-principal mutation gate is frozen; assertion is not consumed}
/createReleaseIntent:
post:
operationId: createReleaseIntent
@@ -0,0 +1,117 @@
# Repository Discovery P0 Runbook
This runbook covers the read-only repository discovery surface and the
governed workflow used to publish a change to it. It does not authorize a
production deployment, a protected-ref update, a tag operation, or a runtime
mode change.
## Safety invariants
Before any operation, record the repository identity and the exact authority
OID. The only repository identity in this contract is `aios/aios`. A floating
ref is resolved once; all subsequent reads in the response are bound to the
returned full `resolved_oid` (or `resolved_from_oid` and `resolved_to_oid` for
a diff) and, for files, the returned `blob_oid`.
The following invariants remain frozen for this P0:
```text
AIOS_GITOPS_WRITES_ENABLED=false
legacy_global_writes_enabled=false
18893 runtime_mode=IMMUTABLE_STAGING
direct bare-repository writes/pushes=forbidden
direct Task DB writes=forbidden
```
Discovery is a read-only operation. It does not acquire a WriterLease and it
does not create a Task. Any implementation change still has to use the formal
Task → TaskPrincipal → TaskLease/claim/heartbeat → WriterLease →
`commitFilesGoverned` path. A request to discovery must never be used as a
write-capability probe.
## Execution-plane preflight
1. Prefer the formal HTTPS control plane at the configured public API/Bridge
endpoint. If raw SSH fails before authentication, preserve that evidence
and do not retry arbitrary usernames, keys, ports, or shell commands.
2. Read runtime, capability, snapshot, and OpenAPI facts. Confirm that the
authority identity and OID are the expected ones, and record whether the
discovery routes are actually deployed on that plane.
3. Classify the available plane as A (network-enabled Codex profile), B
(attested online worker), C (formal HTTPS control plane), or D (existing
executor/coordinator). Select only a supported plane; do not invent a new
SSH or credential path.
4. Treat a readable runtime or a candidate branch as evidence of availability,
not as authorization to write.
The durable discovery record is
`execution-plane-discovery.json`. It must not be included in the governed
source change unless the task explicitly allowlists it.
## Read operations
The canonical Bridge routes are:
| Operation | Method | Required query fields | Optional query fields |
| --- | --- | --- | --- |
| `listRepositoryTree` | `GET /v2/repository/tree` | `repository`, `ref`, `limit` | `prefix`, `depth`, `cursor` |
| `searchRepositorySource` | `GET /v2/repository/search` | `repository`, `ref`, `query`, `max_matches`, `max_bytes` | `path_globs`, `cursor` |
| `getDiffIndex` | `GET /v2/repository/diff-index` | `repository`, `from`, `to`, `limit` | `path_prefix`, `cursor` |
| `readRepositoryFile` | `GET /v2/repository/file` | `repository`, `ref`, `path` | `max_bytes` |
The Actions adapter exposes the corresponding read-only legacy paths under
`/api/repository/*`; the OpenAPI document records each mapping with
`x-aios-legacy-path`.
Every request is bounded. Repository identity is allowlisted, refs are
limited to a branch/tag/full commit form, paths are normalized POSIX-relative
paths, and traversal, option-like paths, shell metacharacters, and ambiguous
refs fail closed. Search skips binary blobs and enforces its cumulative byte
bound. Tree, search, and diff responses are paginated. A cursor is an HMAC-
bound record containing the operation, repository, resolved OID, complete
query, and offset; never edit or manually construct one.
The diff endpoint returns an index only: status, path, optional old path, and
bounded additions/deletions. It never returns a patch body. The file endpoint
returns bounded text plus `content_base64`, `resolved_oid`, and `blob_oid`.
## Publishing an implementation change
1. Preserve the existing dirty worktree. Create a new isolated worktree at
the exact authority OID and use a task-scoped branch.
2. Run discovery unit, pagination, cursor, identity, traversal, binary,
max-byte, immutable-head, and large-diff tests. Run the Bridge/OpenAPI/
Actions drift tests and compile checks.
3. Obtain a formal governed Task and keep its exact task, job, generation,
idempotency, branch, and base-OID identities. Claim and heartbeat through
the supported control plane; do not edit the Task DB or copy a credential.
4. Acquire the Task-bound principal and WriterLease. Submit the allowlisted
source files through `commitFilesGoverned`, then reconcile the Job if the
result is uncertain.
5. Independently read back the authority Git ref, the remote Git API ref, and
the control-plane projection. Completion requires three-way equality to the
governed commit OID, plus the required deployment/acceptance/restart/
official-verification evidence for any release claim.
If the discovery routes are absent from the live OpenAPI/runtime surface, stop
at local validation. Report the implementation and tests as local evidence and
mark governed Task, WriterLease, governed commit, and post-commit readback as
`NOT_VERIFIED`.
## Failure and recovery
- `REPOSITORY_NOT_ALLOWED`, identity mismatch, invalid path/ref, invalid
cursor, or ambiguous ref is a client or security rejection; do not broaden
the allowlist.
- An unavailable authority repository or missing cursor secret is a runtime
configuration failure; do not disable cursor signing or fall back to a
workspace checkout.
- A timeout, oversized output, or large diff is bounded failure. Narrow the
prefix/page or repair the existing supported runtime, never return an
unbounded patch.
- On Windows, a missing `fcntl` means POSIX integration is unverified. Keep
the Python compile and portable unit evidence, then rerun the HTTP and file
locking probes on the supported Linux runtime.
No recovery step in this runbook resets, cleans, overwrites, force-pushes, or
auto-commits a user worktree.
@@ -0,0 +1,32 @@
# Task Lineage Supersession P0 Runbook
## 范围锁定
只允许 source `TASK-SINGLEHOST-TRUSTROOT-P0-20260904`、ordinal `1` 与 child
`TASK-SINGLEHOST-TRUSTROOT-P0-20260904-S1`。每次 live 操作前从 Task API/DB 的只读正式路径
重读 source:必须是 `EXPIRED`、attempt 0、无 claim/lease,并重新计算正式 identity digest。
同时确认 repository、branch、generation、execution digest、external job 与 source metadata
digest 均为 decision 的 exact 值,确认当前 Local Owner Authority 和 global/legacy writes 状态。
## 实施和测试
1. 在专用 staging worktree 完成 controller、API wrapper、migration、child ClaimGuard binding、
ADR/runbook 和 isolated SQLite tests。
2. 覆盖错误 signer/namespace、旧 signer、所有 source CAS 字段漂移、nonce/replay、TTL、
child/payload injection、existing child/lineage、projection/ClaimGuard compatibility 的失败路径。
3. 验证失败后 source 的字节等价快照、child 不存在、没有 lineage/receipt/event 残留。
4. 在 backup/isolated DB 跑 complete dry-run:supersession -> child rev1 -> Initial Projection ->
需要时 Supervisor Projection -> immutable child ClaimGuard binding -> guarded claim attempt 1。
## Staging 激活与 live 操作
仅在测试与 dry-run 全部 PASS 后,将新 API overlay 放在 Task API `PYTHONPATH` 最前并切换到
lineage wrapper;daemon-reload/restart 仅限非生产 Task API/projection/ClaimGuard 服务。验证
health 及固定 supersession route 后,再次重读 live source 并通过当前 Owner authority 签发一份
短 TTL、one-time decision。只调用一次正式 endpoint;不得用 sqlite、maintenance shell、
direct Git ref 或任意签名接口替代。
## 后续和停止条件
成功后只沿 child lineage:Initial Projection、必要的 Supervisor Projection、新 immutable
ClaimGuard binding、worker `p0-task-principal-v15-20260830` 的 guarded claim、governed execution。
@@ -0,0 +1,78 @@
from __future__ import annotations
from pathlib import Path
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
import aios_agent_task_api_lineage_supersession as api
class FakeController:
def __init__(self, db_path, verifier, **kwargs):
self.db_path = db_path
def supersede(self, body):
if body.get("fail"):
raise api.TaskLineageSupersessionError("SOURCE_TASK_EXACT_CAS_MISMATCH", 409)
return {"status": "SUPERSEDED", "idempotent": bool(body.get("replay"))}
def test_endpoint_authenticates_before_controller(monkeypatch):
calls = []
monkeypatch.setattr(api, "TaskLineageSupersessionController", FakeController)
endpoint = api.LineageSupersessionEndpoint(
"unused.sqlite3", object(), lambda headers: calls.append(dict(headers))
)
status, result = endpoint.post({"Authorization": "Bearer control"}, {})
assert status == 201
assert result["status"] == "SUPERSEDED"
assert calls == [{"Authorization": "Bearer control"}]
assert endpoint.post({}, {"replay": True})[0] == 200
assert endpoint.post({}, {"fail": True}) == (
409, {"error": "SOURCE_TASK_EXACT_CAS_MISMATCH"}
)
def test_authentication_failure_prevents_controller(monkeypatch):
called = False
class MustNotRun(FakeController):
def supersede(self, body):
nonlocal called
called = True
monkeypatch.setattr(api, "TaskLineageSupersessionController", MustNotRun)
def deny(_headers):
raise PermissionError("denied")
endpoint = api.LineageSupersessionEndpoint("unused", object(), deny)
with pytest.raises(PermissionError, match="denied"):
endpoint.post({}, {})
assert called is False
def test_runtime_binding_is_lazy_and_fails_closed_when_overlay_missing(monkeypatch):
for name in (
"aios_agent_task_api", "aios_agent_task_api_p0_expired_recovery",
"aios_agent_task_protocol", "aios_owner_approval_verifier",
):
monkeypatch.delitem(sys.modules, name, raising=False)
original = __import__
def blocked(name, *args, **kwargs):
if name == "aios_agent_task_api_p0_expired_recovery":
raise ImportError(name)
return original(name, *args, **kwargs)
monkeypatch.setattr("builtins.__import__", blocked)
with pytest.raises(RuntimeError, match="RUNTIME_OVERLAY_UNAVAILABLE"):
api.build_runtime()
def test_error_status_is_not_inferred_from_untrusted_text():
assert api._status(api.TaskLineageSupersessionError("anything", 403)) == 403
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
from pathlib import Path
import sys
import pytest
APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
if str(APP) not in sys.path:
sys.path.insert(0, str(APP))
from aios_gitops_job_identity import ( # noqa: E402
JobIdentityError,
legacy_task_storage_key,
task_storage_key,
)
def test_collision_resistant_key_separates_historical_lossy_collision() -> None:
first = "TASK/A"
second = "TASK:A"
assert legacy_task_storage_key(first) == legacy_task_storage_key(second) == "TASK-A"
assert task_storage_key(first).startswith("TASK-A-")
assert task_storage_key(second).startswith("TASK-A-")
assert task_storage_key(first) != task_storage_key(second)
@pytest.mark.parametrize("value", ["", " task", "task ", "x" * 513, None])
def test_task_storage_key_rejects_noncanonical_task_identity(value: object) -> None:
with pytest.raises(JobIdentityError, match="TASK_ID_INVALID"):
task_storage_key(value) # type: ignore[arg-type]
@@ -102,6 +102,10 @@ def test_http_route_contract_is_complete():
"/v2/release/execute",
"/v2/release/consistency",
"/v2/release/acceptance",
"/v2/repository/tree",
"/v2/repository/search",
"/v2/repository/diff-index",
"/v2/repository/file",
"/v2/task-admission/envelope",
"/v2/task-admission/authorize",
"/v2/task-admission/status",
+514
View File
@@ -0,0 +1,514 @@
from __future__ import annotations
import json
from pathlib import Path
import subprocess
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from aios_repository_discovery import ( # noqa: E402
RepositoryDiscovery,
RepositoryDiscoveryError,
RepositoryDiscoveryNotFound,
get_diff_index,
list_repository_tree,
read_repository_file,
search_repository_source,
)
REPOSITORY = "aios/aios"
SECRET = b"repository-discovery-test-secret"
def git(*args: str, cwd: Path | None = None) -> str:
return subprocess.run(
["git", *( ["-C", str(cwd)] if cwd is not None else []), *args],
check=True,
capture_output=True,
text=True,
).stdout.strip()
@pytest.fixture()
def repository(tmp_path: Path):
bare = tmp_path / "authoritative.git"
seed = tmp_path / "seed"
git("init", "--bare", str(bare))
git("init", "-b", "main", str(seed))
git("config", "user.name", "Discovery Test", cwd=seed)
git("config", "user.email", "discovery@example.invalid", cwd=seed)
(seed / "README.md").write_text("first immutable snapshot\n", encoding="utf-8")
(seed / "00-large.txt").write_text("x" * 512, encoding="utf-8")
(seed / "mystery" / "jobstore").mkdir(parents=True)
(seed / "mystery" / "jobstore" / "recovery.py").write_text(
"def locate():\n return 'JobStore.for_existing_readonly recovery marker'\n",
encoding="utf-8",
)
for index in range(3):
(seed / "search" / f"match-{index}.py").parent.mkdir(exist_ok=True)
(seed / "search" / f"match-{index}.py").write_text(
f"P0_SEARCH_MARKER line {index}\n", encoding="utf-8"
)
(seed / "binary.bin").write_bytes(b"binary-marker\x00not-source\n")
(seed / "nested" / "a" / "b").mkdir(parents=True)
(seed / "nested" / "a" / "b" / "value.txt").write_text("deep value\n", encoding="utf-8")
git("add", ".", cwd=seed)
git("commit", "-m", "initial", cwd=seed)
main_oid = git("rev-parse", "HEAD", cwd=seed)
git("tag", "v1.0.0", cwd=seed)
git("remote", "add", "origin", str(bare), cwd=seed)
git("push", "origin", "main", "--tags", cwd=seed)
git("--git-dir", str(bare), "symbolic-ref", "HEAD", "refs/heads/main")
git("checkout", "-b", "feature/discovery", cwd=seed)
(seed / "docs").mkdir()
git("mv", "README.md", "docs/README.md", cwd=seed)
for index in range(18):
changed = seed / "zz-diff-index" / f"changed-{index:02}.txt"
changed.parent.mkdir(exist_ok=True)
changed.write_text(f"change {index}\n", encoding="utf-8")
git("add", ".", cwd=seed)
git("commit", "-m", "large indexed diff", cwd=seed)
feature_oid = git("rev-parse", "HEAD", cwd=seed)
git("push", "origin", "feature/discovery", cwd=seed)
git("--git-dir", str(bare), "config", "remote.gitea.url", "https://g.example.invalid/git/aios/aios.git")
git("branch", "ambiguous", feature_oid, cwd=seed)
git("tag", "ambiguous", main_oid, cwd=seed)
git("push", "origin", "refs/heads/ambiguous:refs/heads/ambiguous", cwd=seed)
git("push", "origin", "refs/tags/ambiguous", cwd=seed)
discovery = RepositoryDiscovery(
allowed_repositories=(REPOSITORY,), git_dir=bare, cursor_secret=SECRET,
)
return {"bare": bare, "seed": seed, "main_oid": main_oid, "feature_oid": feature_oid, "discovery": discovery}
def test_exact_allowlist_and_verified_identity_reject_other_repositories(repository):
discovery = repository["discovery"]
with pytest.raises(RepositoryDiscoveryError, match="REPOSITORY_NOT_ALLOWED"):
discovery.list_tree(repository="other/repo", revision="HEAD")
multiple = RepositoryDiscovery(
allowed_repositories=(REPOSITORY, "other/repo"), git_dir=repository["bare"], cursor_secret=SECRET,
)
with pytest.raises(RepositoryDiscoveryError, match="REPOSITORY_IDENTITY_MISMATCH"):
multiple.list_tree(repository="other/repo", revision="HEAD")
def test_head_branch_tag_and_full_oid_resolve_to_one_immutable_commit(repository):
discovery = repository["discovery"]
expected = repository["main_oid"]
assert discovery.resolve_revision(REPOSITORY, "HEAD") == expected
assert discovery.resolve_revision(REPOSITORY, "main") == expected
assert discovery.resolve_revision(REPOSITORY, "v1.0.0") == expected
assert discovery.resolve_revision(REPOSITORY, expected) == expected
for dangerous in ("HEAD~1", "main^{tree}", "--help", "../main", "refs/remotes/origin/main"):
with pytest.raises(RepositoryDiscoveryError, match="INVALID_REVISION"):
discovery.resolve_revision(REPOSITORY, dangerous)
with pytest.raises(RepositoryDiscoveryError, match="AMBIGUOUS_REVISION"):
discovery.resolve_revision(REPOSITORY, "ambiguous")
def test_resolved_oid_remains_immutable_after_floating_head_moves(repository):
discovery = repository["discovery"]
original = discovery.read_file(
repository=REPOSITORY,
revision=repository["main_oid"],
file_path="README.md",
)
assert original["resolved_oid"] == repository["main_oid"]
git(
"--git-dir",
str(repository["bare"]),
"update-ref",
"refs/heads/main",
repository["feature_oid"],
)
moved = discovery.list_tree(
repository=REPOSITORY, revision="main", prefix="docs", limit=10
)
assert moved["resolved_oid"] == repository["feature_oid"]
with pytest.raises(RepositoryDiscoveryNotFound, match="FILE_NOT_FOUND"):
discovery.read_file(
repository=REPOSITORY, revision="main", file_path="README.md"
)
still_original = discovery.read_file(
repository=REPOSITORY,
revision=original["resolved_oid"],
file_path="README.md",
)
pinned_tree = discovery.list_tree(
repository=REPOSITORY, revision=original["resolved_oid"], limit=100
)
pinned_search = discovery.search_source(
repository=REPOSITORY,
revision=original["resolved_oid"],
query="JobStore.for_existing_readonly",
path_globs=("mystery/**/*.py",),
max_matches=5,
max_bytes=4096,
)
pinned_diff = discovery.diff_index(
repository=REPOSITORY,
from_revision=original["resolved_oid"],
to_revision=repository["feature_oid"],
limit=1,
)
assert pinned_tree["resolved_oid"] == repository["main_oid"]
assert any(item["path"] == "README.md" for item in pinned_tree["entries"])
assert pinned_search["resolved_oid"] == repository["main_oid"]
assert pinned_search["matches"]
assert pinned_diff["resolved_from_oid"] == repository["main_oid"]
assert pinned_diff["resolved_to_oid"] == repository["feature_oid"]
assert still_original["resolved_oid"] == repository["main_oid"]
assert still_original["blob_oid"] == original["blob_oid"]
assert still_original["content"] == original["content"]
def test_tree_pagination_cursor_is_signed_and_query_bound(repository):
discovery = repository["discovery"]
first = discovery.list_tree(repository=REPOSITORY, revision="main", limit=2)
assert first["resolved_oid"] == repository["main_oid"]
assert len(first["entries"]) == 2
assert first["truncated"] is True
cursor = first["next_cursor"]
assert isinstance(cursor, str)
second = discovery.list_tree(repository=REPOSITORY, revision=repository["main_oid"], limit=2, cursor=cursor)
assert {item["path"] for item in first["entries"]}.isdisjoint({item["path"] for item in second["entries"]})
with pytest.raises(RepositoryDiscoveryError, match="CURSOR_INVALID"):
discovery.list_tree(repository=REPOSITORY, revision="main", limit=3, cursor=cursor)
tampered_cursor = cursor[:-1] + ("0" if cursor[-1] != "0" else "1")
with pytest.raises(RepositoryDiscoveryError, match="CURSOR_INVALID"):
discovery.list_tree(repository=REPOSITORY, revision="main", limit=2, cursor=tampered_cursor)
@pytest.mark.parametrize("value", ("../secret", "/etc/passwd", "nested/../a", "--output", "nested\\a", ":(glob)**", "nested/*.txt"))
def test_path_traversal_and_git_option_like_paths_are_rejected(repository, value):
discovery = repository["discovery"]
with pytest.raises(RepositoryDiscoveryError, match="PATH_INVALID"):
discovery.list_tree(repository=REPOSITORY, revision="main", prefix=value)
with pytest.raises(RepositoryDiscoveryError, match="PATH_INVALID"):
discovery.read_file(repository=REPOSITORY, revision="main", file_path=value)
def test_source_search_finds_unknown_jobstore_path_with_literal_globs_and_bounds(repository):
discovery = repository["discovery"]
result = discovery.search_source(
repository=REPOSITORY,
revision="main",
query="JobStore.for_existing_readonly",
path_globs=("mystery/**/*.py",),
max_matches=5,
max_bytes=4096,
)
assert result["resolved_oid"] == repository["main_oid"]
assert result["matches"]
match = result["matches"][0]
assert match["path"] == "mystery/jobstore/recovery.py"
assert match["line_start"] == 2
assert match["line_end"] == 2
assert len(match["blob_oid"]) == 40
assert "recovery marker" in match["snippet"]
bounded = discovery.search_source(
repository=REPOSITORY, revision="main", query="x", max_matches=1, max_bytes=8,
)
assert bounded["scanned_bytes"] <= 8
assert bounded["truncated"] is True
def test_source_search_cursor_is_signed_and_binary_blobs_are_skipped(repository):
discovery = repository["discovery"]
first = discovery.search_source(
repository=REPOSITORY,
revision="main",
query="P0_SEARCH_MARKER",
path_globs=("search/*.py",),
max_matches=1,
max_bytes=4096,
)
assert len(first["matches"]) == 1
assert first["truncated"] is True
assert isinstance(first["next_cursor"], str)
second = discovery.search_source(
repository=REPOSITORY,
revision="main",
query="P0_SEARCH_MARKER",
path_globs=("search/*.py",),
max_matches=1,
max_bytes=4096,
cursor=first["next_cursor"],
)
assert second["matches"]
assert first["matches"][0]["path"] != second["matches"][0]["path"]
with pytest.raises(RepositoryDiscoveryError, match="CURSOR_INVALID"):
discovery.search_source(
repository=REPOSITORY,
revision="main",
query="different",
path_globs=("search/*.py",),
max_matches=1,
max_bytes=4096,
cursor=first["next_cursor"],
)
binary = discovery.search_source(
repository=REPOSITORY,
revision="main",
query="binary-marker",
path_globs=("binary.bin",),
max_matches=5,
max_bytes=4096,
)
assert binary["matches"] == []
assert binary["binary_blobs_skipped"] == 1
def test_diff_index_is_paginated_rename_aware_and_never_returns_patch(repository):
discovery = repository["discovery"]
first = discovery.diff_index(
repository=REPOSITORY, from_revision="main", to_revision="feature/discovery", limit=5,
)
assert first["resolved_from_oid"] == repository["main_oid"]
assert first["resolved_to_oid"] == repository["feature_oid"]
assert len(first["entries"]) == 5
assert first["truncated"] is True
assert "patch" not in first
assert all("additions" in item and "deletions" in item for item in first["entries"])
assert any(item["old_path"] == "README.md" and item["path"] == "docs/README.md" for item in first["entries"])
second = discovery.diff_index(
repository=REPOSITORY, from_revision="main", to_revision="feature/discovery", limit=5, cursor=first["next_cursor"],
)
assert {item["path"] for item in first["entries"]}.isdisjoint({item["path"] for item in second["entries"]})
scoped = discovery.diff_index(
repository=REPOSITORY,
from_revision="main",
to_revision="feature/discovery",
prefix="zz-diff-index",
limit=100,
)
assert scoped["entries"]
assert all(item["path"].startswith("zz-diff-index/") for item in scoped["entries"])
assert "resolved_from_oid" in scoped and "resolved_to_oid" in scoped
def test_read_file_returns_resolved_and_blob_identity_with_bounded_content(repository):
discovery = repository["discovery"]
result = discovery.read_file(
repository=REPOSITORY, revision="main", file_path="README.md", max_bytes=5,
)
assert result["resolved_oid"] == repository["main_oid"]
assert len(result["blob_oid"]) == 40
assert result["content"] == "first"
assert result["content_bytes"] == 5
assert result["truncated"] is True
with pytest.raises(RepositoryDiscoveryNotFound, match="FILE_NOT_FOUND"):
discovery.read_file(repository=REPOSITORY, revision="main", file_path="missing.txt")
def test_mapping_adapters_and_module_helpers_return_dicts_with_status_codes(repository):
discovery = repository["discovery"]
tree = discovery.list_repository_tree({"repository": REPOSITORY, "revision": "main", "limit": 2})
assert isinstance(tree, dict)
source = search_repository_source(
{"repository": REPOSITORY, "revision": "main", "query": "recovery marker"}, discovery=discovery,
)
assert source["matches"]
index = get_diff_index(
{"repository": REPOSITORY, "from": "main", "to": "feature/discovery", "limit": 2}, discovery=discovery,
)
assert index["entries"]
blob = read_repository_file(
{"repository": REPOSITORY, "revision": "main", "path": "README.md", "max_bytes": 8}, discovery=discovery,
)
assert blob["path"] == "README.md"
assert list_repository_tree({"repository": REPOSITORY, "revision": "main", "limit": 1}, discovery=discovery)["entries"]
with pytest.raises(RepositoryDiscoveryError) as failure:
discovery.read_repository_file({"repository": REPOSITORY, "revision": "main", "path": "../secret"})
assert str(failure.value) == "REPOSITORY_DISCOVERY_PATH_INVALID"
assert failure.value.status_code == 400
def test_public_request_names_pin_ref_and_return_required_identity_fields(repository):
discovery = repository["discovery"]
tree = discovery.list_repository_tree(
{"repository": REPOSITORY, "ref": "feature/discovery", "limit": 1}
)
assert tree["resolved_oid"] == repository["feature_oid"]
assert {"path", "type", "object_oid"} <= set(tree["entries"][0])
source = discovery.search_repository_source(
{
"repository": REPOSITORY,
"ref": "main",
"query": "P0_SEARCH_MARKER",
"path_globs": ["search/*.py"],
"max_matches": 2,
"max_bytes": 4096,
}
)
assert source["matches"][0]["line_start"] == source["matches"][0]["line_end"] == 1
blob = discovery.read_repository_file(
{"repository": REPOSITORY, "ref": "main", "path": "README.md", "max_bytes": 8}
)
assert blob["content_base64"]
assert len(blob["blob_oid"]) == 40
@pytest.fixture()
def large_diff_repository(tmp_path: Path):
bare = tmp_path / "large.git"
seed = tmp_path / "seed"
git("init", "--bare", str(bare))
git("init", "-b", "main", str(seed))
git("config", "user.name", "Discovery Test", cwd=seed)
git("config", "user.email", "discovery@example.invalid", cwd=seed)
(seed / "README.md").write_text("base\n", encoding="utf-8")
git("add", ".", cwd=seed)
git("commit", "-m", "base", cwd=seed)
main_oid = git("rev-parse", "HEAD", cwd=seed)
git("remote", "add", "origin", str(bare), cwd=seed)
git("push", "origin", "main", cwd=seed)
git("--git-dir", str(bare), "symbolic-ref", "HEAD", "refs/heads/main")
for index in range(18_000):
filename = "long-prefix-" + ("x" * 42) + f"-{index:05}.txt"
(seed / "bulk" / filename).parent.mkdir(exist_ok=True)
(seed / "bulk" / filename).write_text("large diff\n", encoding="utf-8")
git("add", "bulk", cwd=seed)
git("commit", "-m", "large diff", cwd=seed)
feature_oid = git("rev-parse", "HEAD", cwd=seed)
git("push", "origin", "main", cwd=seed)
git("--git-dir", str(bare), "config", "remote.gitea.url", "https://g.example.invalid/git/aios/aios.git")
return {
"bare": bare,
"main_oid": main_oid,
"feature_oid": feature_oid,
"discovery": RepositoryDiscovery(
allowed_repositories=(REPOSITORY,), git_dir=bare, cursor_secret=SECRET
),
}
def test_large_diff_index_is_streamed_and_page_bounded(large_diff_repository):
result = large_diff_repository["discovery"].diff_index(
repository=REPOSITORY,
from_revision=large_diff_repository["main_oid"],
to_revision=large_diff_repository["feature_oid"],
limit=1,
)
assert len(result["entries"]) == 1
assert result["entries"][0]["path"].startswith("bulk/")
assert result["entries"][0]["additions"] == 1
assert result["entries"][0]["deletions"] == 0
assert result["truncated"] is True
def test_bridge_http_discovery_routes_are_get_only_and_do_not_enter_write_handlers(repository):
pytest.importorskip(
"fcntl",
reason="Bridge imports POSIX-only fcntl; Windows integration is unverified",
)
from http.client import HTTPConnection
from http.server import ThreadingHTTPServer
from threading import Thread
from urllib.parse import urlencode
from aios_chatgpt_bridge_v2 import BridgeV2Adapter
from aios_chatgpt_bridge_v2_server import BridgeHandler
discovery = repository["discovery"]
calls: list[str] = []
def invoke(operation: str, method: str):
def handler(request):
calls.append(operation)
return getattr(discovery, method)(request)
return handler
adapter = BridgeV2Adapter(
runtime_status={"deployment_state": "STAGING"},
handlers={
"listRepositoryTree": invoke("listRepositoryTree", "list_repository_tree"),
"searchRepositorySource": invoke(
"searchRepositorySource", "search_repository_source"
),
"getDiffIndex": invoke("getDiffIndex", "get_diff_index"),
"readRepositoryFile": invoke("readRepositoryFile", "read_repository_file"),
},
)
previous_adapter = BridgeHandler.adapter
server = ThreadingHTTPServer(("127.0.0.1", 0), BridgeHandler)
BridgeHandler.adapter = adapter
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
def request(method: str, path: str, body: str | None = None):
connection = HTTPConnection("127.0.0.1", server.server_port, timeout=10)
headers = {"Content-Type": "application/json"} if body is not None else {}
connection.request(method, path, body=body, headers=headers)
response = connection.getresponse()
raw = response.read()
status = response.status
connection.close()
return status, raw
try:
cases = [
(
"/v2/repository/tree",
{"repository": REPOSITORY, "ref": "main", "limit": 2},
),
(
"/v2/repository/search",
{
"repository": REPOSITORY,
"ref": "main",
"query": "recovery marker",
"max_matches": 2,
"max_bytes": 4096,
},
),
(
"/v2/repository/diff-index",
{
"repository": REPOSITORY,
"from": "main",
"to": "feature/discovery",
"limit": 2,
},
),
(
"/v2/repository/file",
{"repository": REPOSITORY, "ref": "main", "path": "README.md"},
),
]
for path, values in cases:
status, raw = request("GET", f"{path}?{urlencode(values)}")
assert status == 200, raw.decode("utf-8", errors="replace")
payload = json.loads(raw.decode("utf-8"))
assert payload["repository"] == REPOSITORY
assert "resolved_oid" in payload or "resolved_to_oid" in payload
assert len(calls) == 4
status, _ = request("POST", "/v2/repository/tree", "{}")
assert status == 405
assert len(calls) == 4
duplicate = (
"/v2/repository/tree?repository=aios%2Faios&ref=main&ref=main&limit=1"
)
status, _ = request("GET", duplicate)
assert status == 400
assert len(calls) == 4
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
BridgeHandler.adapter = previous_adapter
+223
View File
@@ -0,0 +1,223 @@
from __future__ import annotations
import ast
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from aios_chatgpt_bridge_v2 import ( # noqa: E402
BridgeV2Adapter,
BridgeV2Error,
OPERATION_OPTIONAL_FIELDS,
OPERATION_REQUIRED_FIELDS,
READ_ONLY_OPERATIONS,
operation_contracts,
)
DISCOVERY_CONTRACTS = {
"listRepositoryTree": {
"path": "/v2/repository/tree",
"legacy_path": "/api/repository/tree",
},
"searchRepositorySource": {
"path": "/v2/repository/search",
"legacy_path": "/api/repository/search",
},
"getDiffIndex": {
"path": "/v2/repository/diff-index",
"legacy_path": "/api/repository/diff-index",
},
"readRepositoryFile": {
"path": "/v2/repository/file",
"legacy_path": "/api/repository/file",
},
}
def _openapi_path_block(document: str, path: str) -> str:
marker = f" {path}:\n"
start = document.index(marker)
end = document.find("\n /", start + len(marker))
return document[start:] if end < 0 else document[start:end]
def _assignment(source: str, name: str) -> ast.AST:
module = ast.parse(source)
for node in module.body:
if isinstance(node, (ast.Assign, ast.AnnAssign)):
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
if any(isinstance(target, ast.Name) and target.id == name for target in targets):
return node.value
raise AssertionError(f"assignment {name!r} not found")
def _literal_assignment(source: str, name: str):
value = _assignment(source, name)
if isinstance(value, ast.Call) and isinstance(value.func, ast.Name):
assert value.func.id == "frozenset"
assert len(value.args) == 1
return set(ast.literal_eval(value.args[0]))
return ast.literal_eval(value)
def test_discovery_runtime_bridge_and_openapi_contracts_are_aligned():
document = (ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml").read_text(
encoding="utf-8"
)
bridge_source = (ROOT / "active-runtime" / "app" / "aios_chatgpt_bridge_v2_server.py").read_text(
encoding="utf-8"
)
actions_source = (ROOT / "active-runtime" / "app" / "aios_chatgpt_actions_api.py").read_text(
encoding="utf-8"
)
routes = _literal_assignment(bridge_source, "ROUTES")
discovery_actions = _literal_assignment(actions_source, "_DISCOVERY_ACTIONS")
read_only_paths = _literal_assignment(actions_source, "READ_ONLY_GET_PATHS")
runtime = operation_contracts()
for operation, expected in DISCOVERY_CONTRACTS.items():
assert runtime[operation] == {
"method": "GET",
"path": expected["path"],
"required_fields": list(OPERATION_REQUIRED_FIELDS[operation]),
"optional_fields": list(OPERATION_OPTIONAL_FIELDS[operation]),
"read_only": True,
}
assert operation in READ_ONLY_OPERATIONS
assert routes[expected["path"]] == operation
assert expected["legacy_path"] in read_only_paths
assert expected["legacy_path"] in discovery_actions
block = _openapi_path_block(document, expected["path"])
assert block.count(f"operationId: {operation}") == 1
assert " get:\n" in block
assert " security: []" in block
assert " x-openai-isConsequential: false" in block
assert " x-aios-read-only: true" in block
assert " x-aios-bounded: true" in block
assert " x-aios-repository-scoped: true" in block
assert " x-aios-no-shell: true" in block
assert f" x-aios-legacy-path: {expected['legacy_path']}" in block
parameter_names = {
line.split("name:", 1)[1].split(",", 1)[0].strip()
for line in block.splitlines()
if "name:" in line and "in: query" in line
}
assert parameter_names == set(OPERATION_REQUIRED_FIELDS[operation]) | set(
OPERATION_OPTIONAL_FIELDS[operation]
)
for operation in DISCOVERY_CONTRACTS:
assert document.count(f"operationId: {operation}") == 1
class_nodes = [
node
for node in ast.parse(actions_source).body
if isinstance(node, ast.ClassDef) and node.name == "H"
]
assert len(class_nodes) == 1
post_methods = [
node for node in class_nodes[0].body
if isinstance(node, ast.FunctionDef) and node.name == "do_POST"
]
assert post_methods
effective_post = ast.get_source_segment(actions_source, post_methods[-1]) or ""
assert "method_not_allowed" in effective_post
assert "max_num_fields=64" in actions_source
assert "query_too_large" in actions_source
def test_discovery_adapter_rejects_task_principal_and_write_identity_fields():
called = False
def handler(_request):
nonlocal called
called = True
return {}
adapter = BridgeV2Adapter(
runtime_status={}, handlers={"listRepositoryTree": handler}
)
request = {
"repository": "aios/aios",
"ref": "main",
"limit": 1,
"task_id": "must-not-enter-read-route",
"principal": "must-not-enter-read-route",
"writer_lease": "must-not-enter-read-route",
}
with pytest.raises(BridgeV2Error, match="BRIDGE_REQUEST_FIELDS_UNKNOWN"):
adapter.listRepositoryTree(request)
assert called is False
def test_discovery_source_is_read_only_and_staging_freeze_remains_intact():
source = (ROOT / "active-runtime" / "app" / "aios_repository_discovery.py").read_text(
encoding="utf-8"
)
assert '_READ_ONLY_COMMANDS = frozenset({"cat-file", "diff", "ls-tree", "rev-parse"})' in source
assert "shell=True" not in source
assert "os.system" not in source
assert "update-ref" not in source
assert "git push" not in source
unit = (
ROOT
/ "deploy"
/ "systemd"
/ "aios-gitops-model-api-v2r2c1-staging.service"
).read_text(encoding="utf-8")
assert "AIOS_GITOPS_WRITES_ENABLED=false" in unit
assert "AIOS_GITOPS_PORT=18893" in unit
assert "AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=false" in unit
@pytest.mark.parametrize(
("path", "query"),
[
("/api/repository/tree", "repository=aios%2Faios&ref=main&limit=1"),
(
"/api/repository/search",
"repository=aios%2Faios&ref=main&query=needle&max_matches=1&max_bytes=16",
),
(
"/api/repository/diff-index",
"repository=aios%2Faios&from=main&to=main&limit=1",
),
(
"/api/repository/file",
"repository=aios%2Faios&ref=main&path=README.md",
),
],
)
def test_actions_discovery_get_route_dispatches_and_post_is_rejected(
monkeypatch: pytest.MonkeyPatch, path: str, query: str
):
if sys.platform == "win32":
pytest.skip("Actions adapter imports POSIX-only fcntl; Windows integration is unverified")
import aios_chatgpt_actions_api as actions_api
observed: list[tuple[str, object]] = []
def fake_discovery_request(route: str, values: dict[str, list[str]]):
observed.append((route, values))
return {"route": route}
monkeypatch.setattr(actions_api, "repository_discovery_request", fake_discovery_request)
handler = object.__new__(actions_api.H)
responses: list[tuple[object, int]] = []
handler.sendj = lambda value, code=200: responses.append((value, code))
handler.path = f"{path}?{query}"
actions_api.H.do_GET(handler)
assert observed[-1][0] == path
assert responses[-1] == ({"route": path}, 200)
handler.path = path
actions_api.H.do_POST(handler)
assert responses[-1] == ({"error": "method_not_allowed"}, 405)
+315
View File
@@ -0,0 +1,315 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import hashlib
import json
from pathlib import Path
import sqlite3
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
import aios_task_lineage_supersession as sut
NOW = datetime(2026, 9, 5, 4, 0, tzinfo=timezone.utc)
class Verifier:
def verify(self, *, namespace, payload, signature, key_id):
return (
namespace == sut.OWNER_NAMESPACE
and key_id == sut.OWNER_KEY_ID
and signature == "valid-owner-signature"
and json.loads(payload)["signature_namespace"] == sut.OWNER_NAMESPACE
)
def canon(value):
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def formal_identity(row):
value = {
"task_id": row["task_id"],
"goal": row["goal"],
"instructions": row["instructions"],
"success_criteria": json.loads(row["success_criteria_json"]),
"required_capabilities": json.loads(row["required_capabilities_json"]),
"constraints": json.loads(row["constraints_json"]),
"metadata": json.loads(row["metadata_json"]),
"priority": row["priority"],
"max_attempts": row["max_attempts"],
}
return hashlib.sha256(canon(value).encode()).hexdigest()
def create_db(tmp_path):
path = tmp_path / "task.sqlite3"
connection = sqlite3.connect(path)
connection.executescript("""
CREATE TABLE tasks(
task_id TEXT PRIMARY KEY, goal TEXT NOT NULL, instructions TEXT NOT NULL,
success_criteria_json TEXT NOT NULL, required_capabilities_json TEXT NOT NULL,
constraints_json TEXT NOT NULL, metadata_json TEXT NOT NULL, deadline TEXT,
priority INTEGER NOT NULL, max_attempts INTEGER NOT NULL,
attempt INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, claimed_by TEXT,
lease_token_hash TEXT, lease_expires_at TEXT, result_json TEXT,
result_digest TEXT, result_idempotency_key TEXT, verification_json TEXT,
revision INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
);
CREATE TABLE events(
event_id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT, agent_id TEXT,
event_type TEXT NOT NULL, payload_json TEXT NOT NULL, created_at TEXT NOT NULL
);
""")
execution = {
"task_id": sut.SOURCE_TASK_ID,
"repository": sut.REPOSITORY,
"target_branch": sut.TARGET_BRANCH,
"generation": sut.GENERATION,
"execution_digest": sut.EXECUTION_DIGEST,
"external_job_id": sut.EXTERNAL_JOB_ID,
}
connection.execute(
"INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(
sut.SOURCE_TASK_ID, "Build trust root", "Use the governed writer.",
canon([{"id": "complete", "description": "complete"}]),
canon(["task.execute"]), canon({"generation": 1, "production_write": False}),
canon({"execution": execution, "projection": {"already": "historical"}}),
"2026-09-04T00:00:00+00:00", 100, 20, 0, "EXPIRED", None, None, None,
None, None, None, None, 12, "2026-09-04T00:00:00+00:00", "2026-09-04T01:00:00+00:00",
),
)
connection.execute(
"INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) VALUES(?,?,?,?,?)",
(sut.SOURCE_TASK_ID, None, "TASK_CREATED", "{}", "2026-09-04T00:00:00+00:00"),
)
connection.executescript((ROOT / "active-runtime" / "migrations" / "005_task_lineage_supersession.sql").read_text())
connection.commit()
return path
def row(path, task_id=sut.SOURCE_TASK_ID):
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
result = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
connection.close()
return result
def snapshot(path, task_id=sut.SOURCE_TASK_ID):
value = row(path, task_id)
return None if value is None else tuple(value)
def decision(path, **changes):
source = row(path)
metadata_raw = source["metadata_json"]
value = {
"schema": sut.DECISION_SCHEMA,
"decision_id": "decision-singlehost-lineage-s1",
"nonce": "nonce-singlehost-lineage-s1",
"source_task_id": sut.SOURCE_TASK_ID,
"source_task_identity_sha256": formal_identity(source),
"source_metadata_sha256": hashlib.sha256(metadata_raw.encode()).hexdigest(),
"source_expected_status": "EXPIRED",
"source_expected_revision": int(source["revision"]),
"source_expected_attempt": 0,
"source_expected_claimed_by": None,
"source_expected_lease_token_hash": None,
"source_expected_lease_expires_at": None,
"repository": sut.REPOSITORY,
"target_branch": sut.TARGET_BRANCH,
"generation": sut.GENERATION,
"execution_digest": sut.EXECUTION_DIGEST,
"external_job_id": sut.EXTERNAL_JOB_ID,
"child_task_id": sut.CHILD_TASK_ID,
"supersession_ordinal": 1,
"canonical_projection_profile": sut.CANONICAL_PROFILE,
"canonical_projection_version": sut.CANONICAL_PROFILE_VERSION,
"supersession_reason": sut.SUPERSESSION_REASON,
"owner_identity": sut.OWNER_PRINCIPAL,
"signature_namespace": sut.OWNER_NAMESPACE,
"signature_principal": sut.OWNER_PRINCIPAL,
"key_id": sut.OWNER_KEY_ID,
"issued_at": NOW.isoformat(),
"not_before": NOW.isoformat(),
"expires_at": (NOW + timedelta(seconds=90)).isoformat(),
"one_time_use": True,
"signature": "valid-owner-signature",
}
value.update(changes)
return value
def controller(path):
return sut.TaskLineageSupersessionController(
path, Verifier(), clock=lambda: NOW, identity_digest=formal_identity
)
def test_supersession_is_atomic_append_only_and_exact_replay(tmp_path):
path = create_db(tmp_path)
source_before = snapshot(path)
request = decision(path)
result = controller(path).supersede(request)
assert result["status"] == "SUPERSEDED"
assert result["idempotent"] is False
assert snapshot(path) == source_before
child = row(path, sut.CHILD_TASK_ID)
assert (child["status"], child["revision"], child["attempt"]) == ("QUEUED", 1, 0)
assert (child["claimed_by"], child["lease_token_hash"], child["lease_expires_at"]) == (None, None, None)
assert json.loads(child["metadata_json"]) == {}
assert json.loads(child["required_capabilities_json"]) == []
connection = sqlite3.connect(path)
assert connection.execute(
"SELECT event_type FROM events WHERE task_id=?", (sut.CHILD_TASK_ID,)
).fetchall() == [("TASK_CREATED",)]
event = json.loads(connection.execute(
"SELECT payload_json FROM events WHERE task_id=?", (sut.CHILD_TASK_ID,)
).fetchone()[0])
assert event["supersedes_task_id"] == sut.SOURCE_TASK_ID
assert event["owner_supersession_decision_sha256"] == result["owner_decision_sha256"]
stored_decision = json.loads(connection.execute(
"SELECT decision_json FROM task_lineage_supersession_receipts"
).fetchone()[0])
assert stored_decision == request
assert result["owner_signature_key_id"] == sut.OWNER_KEY_ID
with pytest.raises(sqlite3.IntegrityError, match="APPEND_ONLY"):
connection.execute("UPDATE task_lineage_supersessions SET child_task_id='x'")
connection.close()
replay = controller(path).supersede(request)
assert replay["idempotent"] is True
assert replay["owner_decision_sha256"] == result["owner_decision_sha256"]
@pytest.mark.parametrize("field,value,code", [
("signature", "wrong", "DECISION_SIGNATURE_INVALID"),
("signature_namespace", "wrong", "SIGNATURE_NAMESPACE_MISMATCH"),
("signature_principal", "old-owner", "SIGNATURE_PRINCIPAL_MISMATCH"),
("key_id", "SHA256:old", "KEY_ID_MISMATCH"),
("source_task_id", "TASK-other", "SOURCE_TASK_ID_MISMATCH"),
("source_expected_revision", 11, "SOURCE_TASK_EXACT_CAS_MISMATCH"),
("source_metadata_sha256", "0" * 64, "SOURCE_METADATA_DIGEST_MISMATCH"),
("execution_digest", "0" * 64, "EXECUTION_DIGEST_MISMATCH"),
("external_job_id", "job-00000000000000000000000000000000", "EXTERNAL_JOB_ID_MISMATCH"),
("repository", "other/repo", "REPOSITORY_MISMATCH"),
("target_branch", "agent/other", "TARGET_BRANCH_MISMATCH"),
("child_task_id", sut.CHILD_TASK_ID + "-other", "CHILD_TASK_ID_MISMATCH"),
("one_time_use", False, "ONE_TIME_USE_MISMATCH"),
])
def test_signed_scope_mismatch_fails_without_mutation(tmp_path, field, value, code):
path = create_db(tmp_path)
before = snapshot(path)
with pytest.raises(sut.TaskLineageSupersessionError, match=code):
controller(path).supersede(decision(path, **{field: value}))
assert snapshot(path) == before
assert row(path, sut.CHILD_TASK_ID) is None
@pytest.mark.parametrize("column,value", [
("status", "QUEUED"), ("attempt", 1), ("claimed_by", "worker"),
("lease_token_hash", "hash"), ("lease_expires_at", "2099-01-01T00:00:00Z"),
])
def test_live_source_state_drift_rolls_back(tmp_path, column, value):
path = create_db(tmp_path)
request = decision(path)
connection = sqlite3.connect(path)
connection.execute(f"UPDATE tasks SET {column}=? WHERE task_id=?", (value, sut.SOURCE_TASK_ID))
connection.commit()
before = snapshot(path)
with pytest.raises(sut.TaskLineageSupersessionError, match="SOURCE_TASK_EXACT_CAS_MISMATCH"):
controller(path).supersede(request)
assert snapshot(path) == before
assert row(path, sut.CHILD_TASK_ID) is None
def test_child_or_prior_supersession_conflicts_fail_closed(tmp_path):
path = create_db(tmp_path)
request = decision(path)
connection = sqlite3.connect(path)
source = row(path)
values = list(source)
values[0] = sut.CHILD_TASK_ID
connection.execute("INSERT INTO tasks VALUES(" + ",".join("?" for _ in values) + ")", values)
connection.commit()
before = snapshot(path)
with pytest.raises(sut.TaskLineageSupersessionError, match="CHILD_TASK_ALREADY_EXISTS"):
controller(path).supersede(request)
assert snapshot(path) == before
def test_prior_supersession_is_rejected_without_creating_child(tmp_path):
path = create_db(tmp_path)
request = decision(path)
connection = sqlite3.connect(path)
connection.execute(
"INSERT INTO task_lineage_supersessions VALUES(?,?,?,?,?,?,?,?)",
(
sut.SOURCE_TASK_ID, 1, sut.CHILD_TASK_ID,
request["source_task_identity_sha256"], request["source_metadata_sha256"],
"f" * 64, "prior-decision", NOW.isoformat(),
),
)
connection.commit()
before = snapshot(path)
with pytest.raises(sut.TaskLineageSupersessionError, match="SUPERSESSION_ALREADY_EXISTS"):
controller(path).supersede(request)
assert snapshot(path) == before
assert row(path, sut.CHILD_TASK_ID) is None
def test_event_failure_rolls_back_child_lineage_and_receipt(tmp_path):
path = create_db(tmp_path)
request = decision(path)
before = snapshot(path)
connection = sqlite3.connect(path)
connection.execute("""
CREATE TRIGGER reject_child_event BEFORE INSERT ON events
WHEN NEW.task_id = 'TASK-SINGLEHOST-TRUSTROOT-P0-20260904-S1'
BEGIN SELECT RAISE(ABORT, 'TEST_EVENT_FAILURE'); END
""")
connection.commit()
with pytest.raises(sqlite3.IntegrityError, match="TEST_EVENT_FAILURE"):
controller(path).supersede(request)
assert snapshot(path) == before
assert row(path, sut.CHILD_TASK_ID) is None
connection = sqlite3.connect(path)
assert connection.execute("SELECT count(*) FROM task_lineage_supersessions").fetchone()[0] == 0
assert connection.execute("SELECT count(*) FROM task_lineage_supersession_receipts").fetchone()[0] == 0
connection.close()
def test_nonce_reuse_and_changed_decision_are_rejected(tmp_path):
path = create_db(tmp_path)
request = decision(path)
controller(path).supersede(request)
changed = dict(request)
changed["expires_at"] = (NOW + timedelta(seconds=120)).isoformat()
with pytest.raises(sut.TaskLineageSupersessionError, match="DECISION_REPLAY_MISMATCH"):
controller(path).supersede(changed)
@pytest.mark.parametrize("times", [
{"expires_at": (NOW - timedelta(seconds=1)).isoformat()},
{"not_before": (NOW + timedelta(seconds=1)).isoformat()},
{"expires_at": (NOW + timedelta(minutes=6)).isoformat()},
])
def test_decision_time_window_fails_closed(tmp_path, times):
path = create_db(tmp_path)
with pytest.raises(sut.TaskLineageSupersessionError):
controller(path).supersede(decision(path, **times))
assert row(path, sut.CHILD_TASK_ID) is None
def test_arbitrary_child_payload_is_rejected(tmp_path):
path = create_db(tmp_path)
request = decision(path)
request["child_payload"] = {"status": "RUNNING", "revision": 99}
with pytest.raises(sut.TaskLineageSupersessionError, match="DECISION_FIELDS_INVALID"):
controller(path).supersede(request)
assert row(path, sut.CHILD_TASK_ID) is None
@@ -0,0 +1,333 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import sqlite3
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from aios_agent_task_initial_execution_projection import (
InitialExecutionProjectionError,
InitialExecutionProjectionRequest,
InitialExecutionProjectionStore,
)
from aios_online_agent_execution import (
GovernedExecutionBinding,
project_execution_envelope,
)
import aios_task_lineage_supersession as supersession
from aios_task_lineage_supersession_initial_projection import (
PROJECTION_TABLE,
SOURCE_BUSINESS_EXECUTION_DIGEST,
SupersessionChildInitialProjectionStore,
SupersessionChildProjectionIntent,
)
from test_task_lineage_supersession import (
controller,
create_db,
decision,
snapshot,
)
AUTHORIZATION = "owner-supersession-child-initial-projection-20260905"
CAPABILITY = "agent.runtime.openclaw"
BASE_OID = "a" * 40
@pytest.fixture(autouse=True)
def _authorization(monkeypatch):
monkeypatch.setenv(
"AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID",
AUTHORIZATION,
)
def _sha_json(value):
return hashlib.sha256(
json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
).hexdigest()
def _apply_projection_migration(path: Path) -> None:
sql = (
ROOT
/ "active-runtime"
/ "migrations"
/ "006_task_lineage_supersession_initial_projection.sql"
).read_text()
connection = sqlite3.connect(path)
connection.executescript(sql)
connection.commit()
connection.close()
def _superseded_db(tmp_path: Path) -> tuple[Path, tuple, dict]:
path = create_db(tmp_path)
_apply_projection_migration(path)
source_before = snapshot(path)
supersession_receipt = controller(path).supersede(decision(path))
return path, source_before, supersession_receipt
def _intent(**changes):
value = {
"adapter": "openclaw",
"required_capability": CAPABILITY,
"authorization_id": AUTHORIZATION,
}
value.update(changes)
return value
def _resolver(task_id, job_id, capability, adapter, runtime_input):
binding = GovernedExecutionBinding.from_mapping(
{
"task_id": task_id,
"job_id": job_id,
"repository": supersession.REPOSITORY,
"base_oid": BASE_OID,
"expected_branch_oid": BASE_OID,
"target_branch": supersession.TARGET_BRANCH,
"branch_mode": "CREATE_NEW",
"generation": supersession.GENERATION,
"idempotency_key": "singlehost-lineage-s1-execution-envelope-v1",
}
)
return project_execution_envelope(
binding,
{
"action": "agent_runtime",
"adapter": adapter,
"capability_scope": [capability],
"runtime_input": runtime_input,
},
)
def test_real_projector_abi_derives_distinct_child_digest_and_lineage_receipt(tmp_path):
path, source_before, supersession_receipt = _superseded_db(tmp_path)
result = SupersessionChildInitialProjectionStore(path, _resolver).project(
_intent()
)
assert result["idempotent"] is False
assert result["source_business_execution_digest"] == SOURCE_BUSINESS_EXECUTION_DIGEST
assert result["canonical_child_execution_digest"] != SOURCE_BUSINESS_EXECUTION_DIGEST
assert result["owner_supersession_decision_sha256"] == supersession_receipt["owner_decision_sha256"]
assert snapshot(path) == source_before
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
child = connection.execute(
"SELECT * FROM tasks WHERE task_id=?", (supersession.CHILD_TASK_ID,)
).fetchone()
assert (child["status"], child["revision"], child["attempt"]) == (
"QUEUED",
2,
0,
)
assert child["claimed_by"] is None
metadata = json.loads(child["metadata_json"])
execution = metadata["execution"]
assert execution["task_id"] == supersession.CHILD_TASK_ID
assert execution["job_id"] == supersession.EXTERNAL_JOB_ID
assert execution["repository"] == supersession.REPOSITORY
assert execution["target_branch"] == supersession.TARGET_BRANCH
assert execution["generation"] == supersession.GENERATION
assert execution["execution_digest"] == result["canonical_child_execution_digest"]
assert execution["execution_digest"] != supersession.EXECUTION_DIGEST
assert execution["action"] == "agent_runtime"
assert execution["change_set"] == []
assert execution["verification"] == []
assert json.loads(child["required_capabilities_json"]) == [CAPABILITY]
events = connection.execute(
"SELECT event_type,payload_json FROM events WHERE task_id=? ORDER BY event_id",
(supersession.CHILD_TASK_ID,),
).fetchall()
assert [row["event_type"] for row in events] == [
"TASK_CREATED",
"TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED",
]
projected = json.loads(events[-1]["payload_json"])
assert projected["source_business_execution_digest"] == supersession.EXECUTION_DIGEST
assert projected["canonical_child_execution_digest"] == execution["execution_digest"]
assert projected["owner_supersession_decision_sha256"] == supersession_receipt["owner_decision_sha256"]
durable = connection.execute(
f"SELECT * FROM {PROJECTION_TABLE} WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()
assert durable["source_business_execution_digest"] == supersession.EXECUTION_DIGEST
assert durable["canonical_child_execution_digest"] == execution["execution_digest"]
assert durable["expected_revision"] == 1
assert durable["applied_revision"] == 2
with pytest.raises(sqlite3.IntegrityError, match="APPEND_ONLY"):
connection.execute(
f"UPDATE {PROJECTION_TABLE} SET adapter='hermes' WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
)
connection.rollback()
connection.close()
def test_exact_replay_needs_no_bridge_resolution(tmp_path):
path, _, _ = _superseded_db(tmp_path)
first = SupersessionChildInitialProjectionStore(path, _resolver).project(_intent())
def unavailable(*_args):
raise AssertionError("resolver must not run for exact durable replay")
second = SupersessionChildInitialProjectionStore(path, unavailable).project(
_intent()
)
assert second["idempotent"] is True
assert second["canonical_child_execution_digest"] == first["canonical_child_execution_digest"]
def test_generic_initial_projection_cannot_project_a_formal_lineage_child(tmp_path):
path, _, _ = _superseded_db(tmp_path)
execution = _resolver(
supersession.CHILD_TASK_ID,
supersession.EXTERNAL_JOB_ID,
CAPABILITY,
"openclaw",
{
"goal": "Build trust root",
"instructions": "Use the governed writer.",
"success_criteria": [{"id": "complete", "description": "complete"}],
"constraints": {"generation": 1, "production_write": False},
"scope": {"scope_type": "legacy", "tenant_id": None, "project_id": None},
"trace_id": supersession.CHILD_TASK_ID,
},
)
request = InitialExecutionProjectionRequest.from_mapping(
{
"expected_status": "QUEUED",
"expected_revision": 1,
"expected_attempt": 0,
"expected_metadata_sha256": _sha_json({}),
"expected_required_capabilities_sha256": _sha_json([]),
"expected_job_id": supersession.EXTERNAL_JOB_ID,
"expected_execution_digest": execution["execution_digest"],
"required_capability": CAPABILITY,
"adapter": "openclaw",
"authorization_id": AUTHORIZATION,
"idempotency_key": "forbidden-generic-child-projection-v1",
"reason": "must use formal lineage projection endpoint",
}
)
with pytest.raises(
InitialExecutionProjectionError,
match="SUPERSESSION_INITIAL_PROJECTION_ENDPOINT_REQUIRED",
):
InitialExecutionProjectionStore(path, _resolver).project(
supersession.CHILD_TASK_ID, request
)
connection = sqlite3.connect(path)
child = connection.execute(
"SELECT revision,metadata_json FROM tasks WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()
assert child == (1, "{}")
assert connection.execute(
"SELECT count(*) FROM events WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()[0] == 1
connection.close()
def test_intent_rejects_arbitrary_child_payload_and_mismatched_adapter():
with pytest.raises(
InitialExecutionProjectionError,
match="SUPERSESSION_INITIAL_PROJECTION_FIELDS_INVALID",
):
SupersessionChildProjectionIntent.from_mapping(
{**_intent(), "child_payload": {"metadata": {"execution_digest": "x"}}}
)
with pytest.raises(
InitialExecutionProjectionError, match="RUNTIME_CAPABILITY_MISMATCH"
):
SupersessionChildProjectionIntent.from_mapping(
_intent(adapter="hermes")
)
def test_wrong_canonical_binding_fails_without_task_mutation(tmp_path):
path, _, _ = _superseded_db(tmp_path)
def wrong_binding(_task_id, job_id, capability, adapter, runtime_input):
return _resolver(
supersession.SOURCE_TASK_ID,
job_id,
capability,
adapter,
runtime_input,
)
with pytest.raises(
InitialExecutionProjectionError,
match="SUPERSESSION_CANONICAL_EXECUTION_BINDING_MISMATCH",
):
SupersessionChildInitialProjectionStore(path, wrong_binding).project(
_intent()
)
connection = sqlite3.connect(path)
child = connection.execute(
"SELECT revision,metadata_json,required_capabilities_json FROM tasks "
"WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()
assert child == (1, "{}", "[]")
assert connection.execute(f"SELECT count(*) FROM {PROJECTION_TABLE}").fetchone()[0] == 0
connection.close()
def test_projection_receipt_failure_rolls_back_task_event_and_generic_receipt(tmp_path):
path, _, _ = _superseded_db(tmp_path)
connection = sqlite3.connect(path)
connection.execute(
f"""CREATE TRIGGER force_projection_receipt_failure
BEFORE INSERT ON {PROJECTION_TABLE} BEGIN
SELECT RAISE(ABORT, 'FORCED_PROJECTION_RECEIPT_FAILURE');
END"""
)
connection.commit()
connection.close()
with pytest.raises(sqlite3.IntegrityError, match="FORCED_PROJECTION_RECEIPT_FAILURE"):
SupersessionChildInitialProjectionStore(path, _resolver).project(_intent())
connection = sqlite3.connect(path)
child = connection.execute(
"SELECT revision,metadata_json,required_capabilities_json FROM tasks "
"WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()
assert child == (1, "{}", "[]")
assert connection.execute(
"SELECT count(*) FROM events WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()[0] == 1
assert connection.execute(
"SELECT count(*) FROM task_initial_execution_projections WHERE task_id=?",
(supersession.CHILD_TASK_ID,),
).fetchone()[0] == 0
assert connection.execute(f"SELECT count(*) FROM {PROJECTION_TABLE}").fetchone()[0] == 0
connection.close()
def test_projection_migration_is_idempotent(tmp_path):
path = create_db(tmp_path)
_apply_projection_migration(path)
_apply_projection_migration(path)
connection = sqlite3.connect(path)
assert connection.execute(
"SELECT count(*) FROM sqlite_master WHERE type='table' AND name=?",
(PROJECTION_TABLE,),
).fetchone()[0] == 1
connection.close()
@@ -0,0 +1,184 @@
from __future__ import annotations
from dataclasses import asdict
import json
import os
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from aios_single_host_authority import ssh_fingerprint, ssh_public_line
from aios_single_host_operational_authority import (
OperationalAuthorityError,
OperationalOwnerAuthority,
OperationalVerifier,
build_exact_policy,
)
from aios_task_lineage_supersession_owner_authority import (
CHILD_TASK_ID,
EXTERNAL_JOB_IDENTITY,
GENERATION,
REPOSITORY,
SOURCE_BUSINESS_EXECUTION_DIGEST,
SOURCE_TASK_ID,
SUPERSESSION_ORDINAL,
TARGET_BRANCH,
TaskLineageSupersessionOwnerAuthority,
verify_task_lineage_supersession_scope,
)
FIXED_NOW = 2_000_000_000
SOURCE_IDENTITY = "1e6e543443203debc955b4fc5980433dc1b5e403a15a5d5a30f66dad76784bb9"
METADATA_DIGEST = "e68cdae3ea85d2af74058fa435bab5cdba7f47059cbacfe03e0a21a5d034175f"
def _write_key(path: Path, key: Ed25519PrivateKey) -> None:
path.write_bytes(
key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
)
os.chmod(path, 0o600)
@pytest.fixture
def authority_env(tmp_path):
key = Ed25519PrivateKey.generate()
key_path = tmp_path / "owner.key"
_write_key(key_path, key)
task = {
"task_id": SOURCE_TASK_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"base_oid": "91f104ef62a3b688a9259b96e2ca247e0363b564",
"generation": GENERATION,
"execution_plan_sha256": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_identity": EXTERNAL_JOB_IDENTITY,
}
policy = build_exact_policy(task, worker_id="p0-task-principal-v15-20260830")
assert policy.task_identity_sha256 == SOURCE_IDENTITY
registry_path = tmp_path / "registry.json"
registry_path.write_text(
json.dumps(
{
"schema": "aios.single-host-signer-registry.v1",
"authority_generation": 1,
"signers": [
{
"signer_id": "local-owner-authority",
"role": "OWNER_AUTHORITY",
"provider": "local_unix_service",
"fingerprint": ssh_fingerprint(key.public_key()),
"public_key": ssh_public_line(key.public_key(), "owner"),
"socket": "/run/aios/owner-authority.sock",
"secret_ref": "aios-secret://owner",
"status": "ACTIVE",
"generation": 1,
"parent_signer_id": "",
"parent_fingerprint": "",
}
],
},
sort_keys=True,
)
+ "\n"
)
os.chmod(registry_path, 0o644)
return key_path, registry_path, policy
def test_fixed_constructor_uses_real_owner_signature_and_exact_scope(authority_env):
key_path, registry_path, policy = authority_env
tokens = iter(("a" * 32, "b" * 32))
owner = TaskLineageSupersessionOwnerAuthority(
key_path,
policy,
require_root_owner=False,
clock=lambda: FIXED_NOW,
token_hex=lambda _size: next(tokens),
)
artifact = owner.issue_task_lineage_supersession(
source_expected_revision=12,
source_task_identity_sha256=SOURCE_IDENTITY,
source_metadata_sha256=METADATA_DIGEST,
)
payload = OperationalVerifier(
registry_path,
require_root_owner=False,
).owner(artifact)
verify_task_lineage_supersession_scope(payload, policy, now=FIXED_NOW)
assert payload["source_task_id"] == SOURCE_TASK_ID
assert payload["child_task_id"] == CHILD_TASK_ID
assert payload["supersession_ordinal"] == SUPERSESSION_ORDINAL
assert payload["source_business_execution_digest"] == SOURCE_BUSINESS_EXECUTION_DIGEST
assert payload["decision_id"] == "task-lineage-supersession-" + "a" * 32
assert payload["nonce"] == "task-lineage-supersession-nonce-" + "b" * 32
assert (payload["issued_at"], payload["not_before"], payload["expires_at"]) == (
FIXED_NOW,
FIXED_NOW - 5,
FIXED_NOW + 300,
)
def test_old_owner_signer_has_no_lineage_constructor(authority_env):
key_path, _registry_path, policy = authority_env
old_owner = OperationalOwnerAuthority(
key_path,
policy,
require_root_owner=False,
clock=lambda: FIXED_NOW,
)
assert not hasattr(old_owner, "issue_task_lineage_supersession")
@pytest.mark.parametrize(
("updates", "code"),
[
({"source_expected_revision": 0}, "TASK_LINEAGE_SUPERSESSION_REVISION_INVALID"),
({"source_expected_revision": True}, "TASK_LINEAGE_SUPERSESSION_REVISION_INVALID"),
({"source_task_identity_sha256": "0" * 64}, "TASK_LINEAGE_SUPERSESSION_SOURCE_IDENTITY_INVALID"),
({"source_metadata_sha256": "X" * 64}, "TASK_LINEAGE_SUPERSESSION_METADATA_DIGEST_INVALID"),
],
)
def test_dynamic_cas_inputs_fail_closed(authority_env, updates, code):
key_path, _registry_path, policy = authority_env
owner = TaskLineageSupersessionOwnerAuthority(
key_path,
policy,
require_root_owner=False,
clock=lambda: FIXED_NOW,
)
request = {
"source_expected_revision": 12,
"source_task_identity_sha256": SOURCE_IDENTITY,
"source_metadata_sha256": METADATA_DIGEST,
}
request.update(updates)
with pytest.raises(OperationalAuthorityError, match=code):
owner.issue_task_lineage_supersession(**request)
def test_stale_owner_signer_is_rejected_by_current_registry(authority_env):
key_path, registry_path, policy = authority_env
stale_key_path = key_path.parent / "stale-owner.key"
_write_key(stale_key_path, Ed25519PrivateKey.generate())
tokens = iter(("e" * 32, "f" * 32))
stale_owner = TaskLineageSupersessionOwnerAuthority(
stale_key_path,
policy,
require_root_owner=False,
clock=lambda: FIXED_NOW,
token_hex=lambda _size: next(tokens),
)
artifact = stale_owner.issue_task_lineage_supersession(
source_expected_revision=12,
source_task_identity_sha256=SOURCE_IDENTITY,
source_metadata_sha256=METADATA_DIGEST,
)
verifier = OperationalVerifier(registry_path, require_root_owner=False)
with pytest.raises(OperationalAuthorityError, match="SIGNED_ARTIFACT_KEY_ID_INVALID"):
verifier.owner(artifact)
@@ -0,0 +1,253 @@
from __future__ import annotations
from dataclasses import asdict
import json
import os
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from aios_single_host_authority import ssh_fingerprint, ssh_public_line
from aios_single_host_operational_authority import OperationalVerifier, build_exact_policy
from aios_single_host_operational_authority_server import (
OperationalAuthorityService as OldOperationalAuthorityService,
OperationalServiceError,
)
from aios_task_lineage_supersession_owner_authority import (
EXTERNAL_JOB_IDENTITY,
GENERATION,
REPOSITORY,
SOURCE_BUSINESS_EXECUTION_DIGEST,
SOURCE_TASK_ID,
TARGET_BRANCH,
TASK_LINEAGE_SUPERSESSION_OPERATION,
TASK_LINEAGE_SUPERSESSION_SCHEMA,
verify_task_lineage_supersession_scope,
)
from aios_task_lineage_supersession_owner_authority_server import OperationalAuthorityService
FIXED_NOW = 2_000_000_000
SOURCE_IDENTITY = "1e6e543443203debc955b4fc5980433dc1b5e403a15a5d5a30f66dad76784bb9"
METADATA_DIGEST = "e68cdae3ea85d2af74058fa435bab5cdba7f47059cbacfe03e0a21a5d034175f"
def _write_key(path: Path, key: Ed25519PrivateKey) -> None:
path.write_bytes(
key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
)
os.chmod(path, 0o600)
@pytest.fixture
def service_env(tmp_path):
owner_key = Ed25519PrivateKey.generate()
machine_key = Ed25519PrivateKey.generate()
owner_path = tmp_path / "owner.key"
machine_path = tmp_path / "machine.key"
_write_key(owner_path, owner_key)
_write_key(machine_path, machine_key)
task = {
"task_id": SOURCE_TASK_ID,
"repository": REPOSITORY,
"target_branch": TARGET_BRANCH,
"base_oid": "91f104ef62a3b688a9259b96e2ca247e0363b564",
"generation": GENERATION,
"execution_plan_sha256": SOURCE_BUSINESS_EXECUTION_DIGEST,
"external_job_identity": EXTERNAL_JOB_IDENTITY,
}
policy = build_exact_policy(task, worker_id="p0-task-principal-v15-20260830")
assert policy.task_identity_sha256 == SOURCE_IDENTITY
policy_path = tmp_path / "policy.json"
policy_path.write_text(json.dumps(asdict(policy), sort_keys=True) + "\n")
os.chmod(policy_path, 0o600)
registry_path = tmp_path / "registry.json"
registry_path.write_text(
json.dumps(
{
"schema": "aios.single-host-signer-registry.v1",
"authority_generation": 1,
"signers": [
{
"signer_id": "local-owner-authority",
"role": "OWNER_AUTHORITY",
"provider": "local_unix_service",
"fingerprint": ssh_fingerprint(owner_key.public_key()),
"public_key": ssh_public_line(owner_key.public_key(), "owner"),
"socket": "/run/aios/owner-authority.sock",
"secret_ref": "aios-secret://owner",
"status": "ACTIVE",
"generation": 1,
"parent_signer_id": "",
"parent_fingerprint": "",
},
{
"signer_id": "machine-operational",
"role": "MACHINE_OPERATIONAL",
"provider": "local_unix_service",
"fingerprint": ssh_fingerprint(machine_key.public_key()),
"public_key": ssh_public_line(machine_key.public_key(), "machine"),
"socket": "/run/aios/machine-signer.sock",
"secret_ref": "aios-secret://machine",
"status": "ACTIVE",
"generation": 1,
"parent_signer_id": "local-owner-authority",
"parent_fingerprint": ssh_fingerprint(owner_key.public_key()),
},
],
},
sort_keys=True,
)
+ "\n"
)
os.chmod(registry_path, 0o644)
return {
"policy": policy,
"policy_path": policy_path,
"registry_path": registry_path,
"owner_path": owner_path,
"machine_path": machine_path,
"ledger_path": tmp_path / "ledger.sqlite3",
}
def _service(env, *, role="owner", token_hex=None, ledger_path=None):
return OperationalAuthorityService(
role=role,
policy_path=env["policy_path"],
registry_path=env["registry_path"],
owner_key_path=env["owner_path"],
machine_key_path=env["machine_path"],
socket_path=env["policy_path"].parent / (role + ".sock"),
ledger_path=ledger_path or env["ledger_path"],
allowed_uids=(1000,),
clock=lambda: FIXED_NOW,
require_root_owner=False,
token_hex=token_hex,
)
def _lineage_request(**updates):
payload = {
"source_expected_revision": 12,
"source_task_identity_sha256": SOURCE_IDENTITY,
"source_metadata_sha256": METADATA_DIGEST,
}
payload.update(updates)
return {"operation": TASK_LINEAGE_SUPERSESSION_OPERATION, "payload": payload}
def test_fixed_operation_issues_once_and_persists_rejection_across_restart(service_env):
tokens = iter(("c" * 32, "d" * 32))
first = _service(service_env, token_hex=lambda _size: next(tokens))
result = first.handle(_lineage_request(), peer_uid=1000)
artifact = result["artifact"]
payload = OperationalVerifier(
service_env["registry_path"],
require_root_owner=False,
).owner(artifact)
verify_task_lineage_supersession_scope(payload, service_env["policy"], now=FIXED_NOW)
assert payload["schema"] == TASK_LINEAGE_SUPERSESSION_SCHEMA
restarted = _service(service_env)
with pytest.raises(OperationalServiceError, match="TASK_LINEAGE_SUPERSESSION_ALREADY_ISSUED"):
restarted.handle(_lineage_request(), peer_uid=1000)
@pytest.mark.parametrize(
"override",
[
{"payload": {}},
{"message": "sign this"},
{"bytes": "00"},
{"schema": "aios.other.v1"},
{"source_task_id": "TASK-other"},
{"child_task_id": SOURCE_TASK_ID + "-S2"},
{"supersession_ordinal": 2},
{"repository": "other/repo"},
{"target_branch": "agent/other"},
{"generation": 2},
{"source_business_execution_digest": "0" * 64},
{"external_job_identity": "job-00000000000000000000000000000000"},
{"owner_namespace": "other-owner"},
{"one_time_use": False},
{"decision_id": "caller-id"},
{"nonce": "caller-nonce"},
{"issued_at": FIXED_NOW},
{"not_before": FIXED_NOW},
{"expires_at": FIXED_NOW + 60},
],
)
def test_all_identity_and_generic_signing_overrides_are_rejected(service_env, override):
service = _service(service_env)
request = _lineage_request(**override)
with pytest.raises(OperationalServiceError, match="REQUEST_FIELDS_INVALID"):
service.handle(request, peer_uid=1000)
@pytest.mark.parametrize("operation", ["sign", "sign_bytes", "sign_payload", "issue_signature"])
def test_generic_signing_operations_remain_forbidden(service_env, operation):
service = _service(service_env)
with pytest.raises(OperationalServiceError, match="OWNER_OPERATION_FORBIDDEN"):
service.handle({"operation": operation, "payload": {}}, peer_uid=1000)
def test_old_server_rejects_lineage_operation(service_env):
old = OldOperationalAuthorityService(
role="owner",
policy_path=service_env["policy_path"],
registry_path=service_env["registry_path"],
owner_key_path=service_env["owner_path"],
machine_key_path=service_env["machine_path"],
socket_path=service_env["policy_path"].parent / "old.sock",
ledger_path=service_env["policy_path"].parent / "old-ledger.sqlite3",
allowed_uids=(1000,),
clock=lambda: FIXED_NOW,
require_root_owner=False,
)
with pytest.raises(OperationalServiceError, match="OWNER_OPERATION_FORBIDDEN"):
old.handle(_lineage_request(), peer_uid=1000)
def test_existing_delegation_and_recovery_operations_are_unchanged(service_env):
service = _service(service_env)
delegation_request = {
"operation": "issue_operational_delegation",
"payload": {
"delegation_id": "opdel-lineage-regression",
"nonce": "nonce-lineage-regression-delegation",
"ttl_seconds": 3600,
},
}
delegation = service.handle(delegation_request, peer_uid=1000)["artifact"]
recovery_request = {
"operation": "issue_expired_task_recovery",
"payload": {
"delegation": delegation,
"decision_id": "recovery-lineage-regression",
"nonce": "nonce-lineage-regression-recovery",
"expected_revision": 12,
"expected_attempt": 0,
"original_deadline": FIXED_NOW - 100,
"new_deadline": FIXED_NOW + 3600,
"ttl_seconds": 120,
},
}
recovery = service.handle(recovery_request, peer_uid=1000)["artifact"]
verifier = OperationalVerifier(service_env["registry_path"], require_root_owner=False)
assert verifier.owner(delegation)["schema"] == "aios.owner-operational-delegation.v1"
assert verifier.owner(recovery)["schema"] == "aios.expired-task-recovery-decision.v1"
def test_lineage_operation_is_owner_only(service_env):
service = _service(
service_env,
role="machine",
ledger_path=service_env["policy_path"].parent / "machine-ledger.sqlite3",
)
with pytest.raises(OperationalServiceError, match="MACHINE_OPERATION_FORBIDDEN"):
service.handle(_lineage_request(), peer_uid=1000)
@@ -0,0 +1,191 @@
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import sqlite3
import sys
import types
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
import aios_task_claim_guard_binding_v2 as guard
from aios_online_agent_execution import GovernedExecutionBinding, project_execution_envelope
from aios_task_lineage_supersession_initial_projection import (
SOURCE_BUSINESS_EXECUTION_DIGEST,
SupersessionChildInitialProjectionStore,
)
from test_task_lineage_supersession import create_db, controller, decision
@pytest.fixture(autouse=True)
def projection_authorization(monkeypatch):
monkeypatch.setenv(
"AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID",
"owner-supersession-child-initial-projection-20260905",
)
@dataclass
class Policy:
task_id: str
repository: str
target_branch: str
generation: int
execution_plan_sha256: str
external_job_identity: str
allowed_worker_ids: tuple[str, ...]
base_oid: str
digest: str
def validate(self): return None
def apply_projection_migration(path):
sql = (
ROOT / "active-runtime" / "migrations"
/ "006_task_lineage_supersession_initial_projection.sql"
).read_text()
connection = sqlite3.connect(path)
connection.executescript(sql)
connection.commit()
connection.close()
def resolver(task_id, job_id, capability, adapter, runtime_input):
binding = GovernedExecutionBinding.from_mapping({
"task_id": task_id, "job_id": job_id, "repository": guard.REPOSITORY,
"base_oid": "a" * 40, "expected_branch_oid": "a" * 40,
"target_branch": guard.TARGET_BRANCH, "branch_mode": "CREATE_NEW",
"generation": guard.GENERATION,
"idempotency_key": "singlehost-lineage-s1-guard-test-v1",
})
return project_execution_envelope(binding, {
"action": "agent_runtime", "adapter": adapter,
"capability_scope": [capability], "runtime_input": runtime_input,
})
def projected(path):
apply_projection_migration(path)
result = SupersessionChildInitialProjectionStore(path, resolver).project({
"adapter": "openclaw",
"required_capability": "agent.runtime.openclaw",
"authorization_id": "owner-supersession-child-initial-projection-20260905",
})
connection = sqlite3.connect(path)
return connection, result
def policy_value(connection, **changes):
canonical_digest = connection.execute(
"SELECT canonical_child_execution_digest FROM "
"task_lineage_supersession_initial_projections WHERE task_id=?",
(guard.CHILD_TASK_ID,),
).fetchone()[0]
value = {
"task_id": guard.CHILD_TASK_ID, "repository": guard.REPOSITORY,
"target_branch": guard.TARGET_BRANCH, "generation": guard.GENERATION,
"execution_plan_sha256": canonical_digest,
"external_job_identity": guard.EXTERNAL_JOB_ID,
"allowed_worker_ids": [guard.WORKER_ID], "base_oid": "a" * 40,
}
value.update(changes)
return value
def factory(value):
digest = hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
return Policy(
task_id=value["task_id"], repository=value["repository"],
target_branch=value["target_branch"], generation=value["generation"],
execution_plan_sha256=value["execution_plan_sha256"],
external_job_identity=value["external_job_identity"],
allowed_worker_ids=tuple(value["allowed_worker_ids"]), base_oid=value["base_oid"], digest=digest,
)
def install(connection, **changes):
return guard.install_child_binding(
connection, policy=policy_value(connection, **changes), policy_factory=factory,
runtime_id="runtime-v15", runtime_observation_digest="b" * 64,
task_contract_version="aios.agent-task.v1", now="2026-09-05T04:02:00Z",
)
def test_binding_requires_canonical_projection(tmp_path):
path = create_db(tmp_path)
controller(path).supersede(decision(path))
apply_projection_migration(path)
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
with pytest.raises(
guard.ClaimGuardBindingError,
match="FORMAL_CHILD_INITIAL_PROJECTION_REQUIRED",
):
guard.install_child_binding(
connection, policy={}, runtime_id="runtime-v15",
runtime_observation_digest="b" * 64,
task_contract_version="aios.agent-task.v1", policy_factory=factory,
)
assert connection.execute("SELECT count(*) FROM task_claim_guard_bindings_v2").fetchone()[0] == 0
connection.close()
def test_binding_is_server_derived_append_only_and_exact(tmp_path, monkeypatch):
path = create_db(tmp_path)
supersession = controller(path).supersede(decision(path))
connection, projection = projected(path)
connection.row_factory = sqlite3.Row
result = install(connection)
connection.commit()
assert result["status"] == "BOUND"
assert result["worker_id"] == guard.WORKER_ID
assert result["source_business_execution_digest"] == SOURCE_BUSINESS_EXECUTION_DIGEST
assert result["canonical_child_execution_digest"] == projection["canonical_child_execution_digest"]
assert result["canonical_child_execution_digest"] != result["source_business_execution_digest"]
assert result["owner_supersession_decision_sha256"] == supersession["owner_decision_sha256"]
assert install(connection)["idempotent"] is True
with pytest.raises(sqlite3.IntegrityError, match="APPEND_ONLY"):
connection.execute("UPDATE task_claim_guard_bindings_v2 SET worker_id='other'")
connection.rollback()
ran = []
monkeypatch.setattr(guard, "_run_child_guard", lambda conn, binding, policy, agent: ran.append((binding, agent)))
guard.enforce_claim_guard(connection, task_id=guard.CHILD_TASK_ID, agent_id=guard.WORKER_ID)
assert ran[0][0]["expected_revision"] == 2
with pytest.raises(guard.ClaimGuardBindingError, match="WORKER_ID_MISMATCH"):
guard.enforce_claim_guard(connection, task_id=guard.CHILD_TASK_ID, agent_id="other")
connection.close()
def test_binding_rejects_policy_widening_and_projection_drift(tmp_path):
path = create_db(tmp_path)
controller(path).supersede(decision(path))
connection, projection = projected(path)
connection.row_factory = sqlite3.Row
with pytest.raises(guard.ClaimGuardBindingError, match="POLICY_WORKER_MISMATCH"):
install(connection, allowed_worker_ids=[guard.WORKER_ID, "other"])
assert connection.execute("SELECT count(*) FROM task_claim_guard_bindings_v2").fetchone()[0] == 0
with pytest.raises(
guard.ClaimGuardBindingError, match="OPERATIONAL_POLICY_SCOPE_MISMATCH"
):
install(connection, execution_plan_sha256=SOURCE_BUSINESS_EXECUTION_DIGEST)
assert projection["canonical_child_execution_digest"] != SOURCE_BUSINESS_EXECUTION_DIGEST
install(connection)
connection.commit()
connection.execute("UPDATE tasks SET revision=3 WHERE task_id=?", (guard.CHILD_TASK_ID,))
connection.commit()
with pytest.raises(guard.ClaimGuardBindingError, match="TASK_CAS_MISMATCH"):
guard.enforce_claim_guard(connection, task_id=guard.CHILD_TASK_ID, agent_id=guard.WORKER_ID)
connection.close()
def test_exact_two_task_dispatch(tmp_path, monkeypatch):
calls = []
fake = types.SimpleNamespace(enforce_claim_guard=lambda *a, **kw: calls.append(kw))
monkeypatch.setitem(sys.modules, "aios_singlehost_claim_guard_stage1", fake)
guard.enforce_claim_guard(object(), task_id=guard.SOURCE_TASK_ID, agent_id="legacy")
guard.enforce_claim_guard(object(), task_id="TASK-unrelated", agent_id="any")
assert calls == [{"task_id": guard.SOURCE_TASK_ID, "agent_id": "legacy"}]