feat(gitops): add continuation lease and durable job boundary
This commit is contained in:
chatgpt-final-release-writer
@@ -0,0 +1,95 @@
|
||||
"""Read-only validation of recovered campaign authorization evidence."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import stat
|
||||
from typing import Any
|
||||
|
||||
|
||||
class RecoveryEvidenceError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def _digest(path: Path) -> str:
|
||||
h = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
h.update(block)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def _secure_json(path: Path) -> dict[str, Any]:
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_MISSING")
|
||||
info = path.stat(follow_symlinks=False)
|
||||
if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o022:
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_PERMISSIONS")
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_INVALID") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_OBJECT_REQUIRED")
|
||||
return value
|
||||
|
||||
|
||||
def load_recovered_authorization(path: str | Path) -> dict[str, Any]:
|
||||
"""Verify the recovery receipt, sidecar, and all referenced immutable files."""
|
||||
receipt_path = Path(path)
|
||||
receipt = _secure_json(receipt_path)
|
||||
sidecar = receipt_path.with_suffix(receipt_path.suffix + ".sha256")
|
||||
if sidecar.is_symlink() or not sidecar.is_file():
|
||||
raise RecoveryEvidenceError("RECOVERY_DIGEST_SIDECAR_MISSING")
|
||||
expected = sidecar.read_text(encoding="ascii").split()[0]
|
||||
actual = _digest(receipt_path)
|
||||
if expected != actual:
|
||||
raise RecoveryEvidenceError("RECOVERY_DIGEST_MISMATCH")
|
||||
required = {
|
||||
"schema",
|
||||
"original_event_id",
|
||||
"original_authorization_digest",
|
||||
"signed_grant_id",
|
||||
"signed_grant_digest",
|
||||
"task_id",
|
||||
"repository",
|
||||
"original_base_oid",
|
||||
"branch",
|
||||
"recovered_from_existing_evidence",
|
||||
"new_owner_interaction",
|
||||
"state",
|
||||
"production_side_effect_executed",
|
||||
"global_writes_enabled",
|
||||
"evidence_files",
|
||||
}
|
||||
if set(receipt) != required or receipt["schema"] != "aios.owner-authorization-recovery.v1":
|
||||
raise RecoveryEvidenceError("RECOVERY_SCHEMA_INVALID")
|
||||
if receipt["recovered_from_existing_evidence"] is not True or receipt["new_owner_interaction"] is not False:
|
||||
raise RecoveryEvidenceError("RECOVERY_NOT_APPEND_ONLY")
|
||||
if receipt["production_side_effect_executed"] is not False or receipt["global_writes_enabled"] is not False:
|
||||
raise RecoveryEvidenceError("RECOVERY_SCOPE_INVALID")
|
||||
if receipt["state"] != "RECOVERED_PENDING_CONTINUATION_LEASE":
|
||||
raise RecoveryEvidenceError("RECOVERY_STATE_INVALID")
|
||||
evidence = receipt["evidence_files"]
|
||||
if not isinstance(evidence, list) or not evidence:
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_LIST_INVALID")
|
||||
for item in evidence:
|
||||
if not isinstance(item, dict) or set(item) != {"path", "sha256"}:
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_ENTRY_INVALID")
|
||||
target = Path(item["path"])
|
||||
if _digest(target) != item["sha256"]:
|
||||
raise RecoveryEvidenceError("RECOVERY_EVIDENCE_DIGEST_MISMATCH")
|
||||
return {
|
||||
"schema": receipt["schema"],
|
||||
"state": receipt["state"],
|
||||
"original_event_id": receipt["original_event_id"],
|
||||
"signed_grant_id": receipt["signed_grant_id"],
|
||||
"task_id": receipt["task_id"],
|
||||
"repository": receipt["repository"],
|
||||
"original_base_oid": receipt["original_base_oid"],
|
||||
"branch": receipt["branch"],
|
||||
"digest": actual,
|
||||
"new_owner_interaction": False,
|
||||
}
|
||||
@@ -33,6 +33,11 @@ from silent_release_hardening import (
|
||||
SignedReleasePlanApproval,
|
||||
SilentReleaseActionService,
|
||||
)
|
||||
try:
|
||||
from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
|
||||
except ModuleNotFoundError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
|
||||
try:
|
||||
from authoritative_git import (
|
||||
AuthoritativeGitError,
|
||||
@@ -120,6 +125,12 @@ def reject_non_read_http(handler, method: str, path: str) -> bool:
|
||||
return False
|
||||
if path in SILENT_RELEASE_POST_PATHS:
|
||||
return False
|
||||
if (
|
||||
method == 'POST'
|
||||
and path in ('/api/file', '/api/commit-file')
|
||||
and os.getenv('AIOS_GITOPS_JOB_CONTROLLER_ENABLED', 'false').lower() in {'1', 'true', 'yes', 'on'}
|
||||
):
|
||||
return False
|
||||
decision = evaluate_git_write(WS)
|
||||
if decision.allowed:
|
||||
return False
|
||||
@@ -666,7 +677,39 @@ class H(BaseHTTPRequestHandler):
|
||||
def file(self,data):
|
||||
a=self.need('file:write');
|
||||
if not a: return
|
||||
b=checkout(data.get('branch'), data.get('base','main'), bool(data.get('create_branch'))); r=write_file(data.get('path'), data.get('content','')); c=commit([r], data.get('message') or f'chore(chatgpt): update {r}'); audit(a['account'],'file',r,{'branch':b,'commit':c}); self.sendj({'status':'committed','branch':b,'path':r,'commit':c})
|
||||
if os.getenv('AIOS_GITOPS_JOB_CONTROLLER_ENABLED', 'false').lower() not in {'1','true','yes','on'}:
|
||||
return self.sendj({'error':'DURABLE_JOB_CONTROLLER_UNAVAILABLE','retryable':True},503)
|
||||
try:
|
||||
path = relpath(str(data.get('path') or ''))
|
||||
branch = str(data.get('branch') or '')
|
||||
expected = str(data.get('expected_branch_oid') or '')
|
||||
campaign_id = str(data.get('campaign_id') or os.getenv('AIOS_GITOPS_CAMPAIGN_ID') or '')
|
||||
supplied_digest = str(data.get('payload_digest') or '')
|
||||
content = data.get('content')
|
||||
payload_digest = supplied_digest or hashlib.sha256(str(content or '').encode('utf-8')).hexdigest()
|
||||
if not campaign_id or not branch or not expected:
|
||||
return self.sendj({'error':'JOB_BINDING_REQUIRED','retryable':False},409)
|
||||
request = {
|
||||
'campaign_id': campaign_id,
|
||||
'repository': REPOSITORY_IDENTITY or 'aios/aios',
|
||||
'operation': 'commit-file',
|
||||
'branch': branch,
|
||||
'expected_branch_oid': expected,
|
||||
'payload_digest': payload_digest,
|
||||
# The controller persists only the binding and digest. Raw
|
||||
# content is never copied into durable state or logs.
|
||||
}
|
||||
receipt = GitOpsJobController(
|
||||
JobStore(Path(os.getenv('AIOS_GITOPS_JOB_ROOT', '/var/lib/aios-gitops/jobs')))
|
||||
).submit(request)
|
||||
return self.sendj({
|
||||
'job_id': receipt.job_id,
|
||||
'state': receipt.state.value,
|
||||
'side_effect_state': receipt.side_effect_state,
|
||||
'retryable': receipt.http_status == 503,
|
||||
}, receipt.http_status)
|
||||
except JobError as exc:
|
||||
return self.sendj({'error':str(exc),'retryable':False},409)
|
||||
def commit(self,data):
|
||||
a=self.need('commit:create');
|
||||
if not a: return
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Campaign-aware continuation leases for an existing Git branch.
|
||||
|
||||
This module is deliberately transport-neutral. It validates branch lineage
|
||||
and hands ref mutation to an injected compare-and-swap broker; it does not run
|
||||
Git, inspect credentials, or create a branch itself.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import re
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
TARGET_BRANCH = "agent/autonomous-release-campaign-v1-20260807"
|
||||
LINEAGE_ANCHOR = "69b45e81dcd2c3707c367003b08c55fd361faae8"
|
||||
CAMPAIGN_HEAD = "bba675fab3b7e0c5fba7066720252b3c91b9f20d"
|
||||
|
||||
|
||||
class ContinuationError(RuntimeError):
|
||||
"""Stable fail-closed error for lease or branch binding violations."""
|
||||
|
||||
|
||||
class BranchProbe(Protocol):
|
||||
def branch_exists(self, branch: str) -> bool: ...
|
||||
def branch_head(self, branch: str) -> str: ...
|
||||
def is_ancestor(self, older: str, newer: str) -> bool: ...
|
||||
|
||||
|
||||
class RefCAS(Protocol):
|
||||
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...
|
||||
|
||||
|
||||
def _oid(value: str, field: str) -> str:
|
||||
if not isinstance(value, str) or not OID_RE.fullmatch(value):
|
||||
raise ContinuationError(f"{field}_invalid")
|
||||
return value
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class WriterLeaseRequestV2:
|
||||
task_id: str
|
||||
repository: str
|
||||
branch_mode: str
|
||||
base_oid: str
|
||||
expected_branch_oid: str
|
||||
target_branch: str
|
||||
generation: int
|
||||
lease_seconds: int = 900
|
||||
|
||||
schema = "aios.task-writer-lease-request.v2"
|
||||
|
||||
def validate(self, probe: BranchProbe) -> None:
|
||||
if not self.task_id or self.repository != "aios/aios":
|
||||
raise ContinuationError("request_identity_invalid")
|
||||
if self.branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
|
||||
raise ContinuationError("branch_mode_invalid")
|
||||
_oid(self.base_oid, "base_oid")
|
||||
_oid(self.expected_branch_oid, "expected_branch_oid")
|
||||
if self.target_branch in {"main", "master"} or self.target_branch.startswith(("refs/", "tags/")):
|
||||
raise ContinuationError("protected_branch_rejected")
|
||||
if not isinstance(self.generation, int) or isinstance(self.generation, bool) or self.generation < 1:
|
||||
raise ContinuationError("generation_invalid")
|
||||
if not isinstance(self.lease_seconds, int) or not 1 <= self.lease_seconds <= 86_400:
|
||||
raise ContinuationError("lease_duration_invalid")
|
||||
if self.branch_mode == "CONTINUE_EXISTING":
|
||||
if not probe.branch_exists(self.target_branch):
|
||||
raise ContinuationError("EXISTING_BRANCH_MISSING")
|
||||
actual = _oid(probe.branch_head(self.target_branch), "actual_branch_oid")
|
||||
if actual != self.expected_branch_oid:
|
||||
raise ContinuationError("EXISTING_BRANCH_OID_CONFLICT")
|
||||
if not probe.is_ancestor(self.base_oid, actual):
|
||||
raise ContinuationError("BASE_NOT_ANCESTOR_OF_BRANCH")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CampaignWriterLease:
|
||||
task_id: str
|
||||
repository: str
|
||||
branch: str
|
||||
lineage_anchor_oid: str
|
||||
current_head_oid: str
|
||||
generation: int
|
||||
lease_expires_at: datetime
|
||||
append_only: bool = True
|
||||
renewable: bool = True
|
||||
multi_commit: bool = True
|
||||
recoverable: bool = True
|
||||
state: str = "ACTIVE"
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_oid(self.lineage_anchor_oid, "lineage_anchor_oid")
|
||||
_oid(self.current_head_oid, "current_head_oid")
|
||||
if not all((self.append_only, self.renewable, self.multi_commit, self.recoverable)):
|
||||
raise ContinuationError("lease_capability_downgrade")
|
||||
|
||||
|
||||
class ContinuationLeaseManager:
|
||||
"""State machine for a multi-commit lease with CAS-only ref updates."""
|
||||
|
||||
def __init__(self, probe: BranchProbe, refs: RefCAS, *, now=datetime.now) -> None:
|
||||
self.probe = probe
|
||||
self.refs = refs
|
||||
self.now = now
|
||||
|
||||
def open(self, request: WriterLeaseRequestV2) -> CampaignWriterLease:
|
||||
request.validate(self.probe)
|
||||
if request.branch_mode != "CONTINUE_EXISTING":
|
||||
raise ContinuationError("CREATE_NEW_REQUIRES_SEPARATE_PROVISIONER")
|
||||
return CampaignWriterLease(
|
||||
task_id=request.task_id,
|
||||
repository=request.repository,
|
||||
branch=request.target_branch,
|
||||
lineage_anchor_oid=request.base_oid,
|
||||
current_head_oid=request.expected_branch_oid,
|
||||
generation=request.generation,
|
||||
lease_expires_at=self.now(timezone.utc) + timedelta(seconds=request.lease_seconds),
|
||||
)
|
||||
|
||||
def renew(self, lease: CampaignWriterLease, seconds: int) -> CampaignWriterLease:
|
||||
if lease.state != "ACTIVE" or not lease.renewable or seconds < 1:
|
||||
raise ContinuationError("lease_not_renewable")
|
||||
return replace(lease, lease_expires_at=self.now(timezone.utc) + timedelta(seconds=seconds))
|
||||
|
||||
def append_commit(self, lease: CampaignWriterLease, new_head_oid: str) -> CampaignWriterLease:
|
||||
if lease.state != "ACTIVE":
|
||||
raise ContinuationError("lease_not_active")
|
||||
_oid(new_head_oid, "new_head_oid")
|
||||
try:
|
||||
actual = self.refs.compare_and_swap_branch(
|
||||
lease.branch, lease.current_head_oid, new_head_oid
|
||||
)
|
||||
except Exception as exc:
|
||||
raise ContinuationError("BRANCH_LEASE_STALE") from exc
|
||||
if actual != new_head_oid:
|
||||
raise ContinuationError("BRANCH_LEASE_STALE")
|
||||
return replace(lease, current_head_oid=new_head_oid)
|
||||
|
||||
def supersede(self, lease: CampaignWriterLease) -> CampaignWriterLease:
|
||||
if lease.state not in {"ACTIVE", "RECOVERABLE_FAILURE", "RECONCILIATION_REQUIRED"}:
|
||||
raise ContinuationError("lease_not_supersedable")
|
||||
return replace(lease, generation=lease.generation + 1, state="SUPERSEDED")
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Non-Owner Gitea machine capability boundary.
|
||||
|
||||
Only an opaque credential projection reference is accepted. The broker
|
||||
delegates network work to an injected transport and never accepts a plaintext
|
||||
token, Owner credential, force push, admin operation, or direct main write.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Mapping, Protocol
|
||||
|
||||
|
||||
class MachineBrokerError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class GiteaTransport(Protocol):
|
||||
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str: ...
|
||||
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]: ...
|
||||
def publish_status(self, branch: str, state: str, description: str) -> Mapping[str, Any]: ...
|
||||
def protected_merge(self, pr_number: int) -> Mapping[str, Any]: ...
|
||||
def read_branch(self, branch: str) -> str | None: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MachineCredentialProjection:
|
||||
reference: str
|
||||
account: str
|
||||
owner_account: str = "aiman"
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
lowered = self.reference.lower()
|
||||
if not self.reference or "token" in lowered or "password" in lowered:
|
||||
raise MachineBrokerError("PLAINTEXT_CREDENTIAL_REJECTED")
|
||||
if not self.account or self.account == self.owner_account:
|
||||
raise MachineBrokerError("OWNER_MACHINE_IDENTITY_REJECTED")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CampaignCapabilityReceipt:
|
||||
campaign_id: str
|
||||
task_id: str
|
||||
repository: str
|
||||
branch: str
|
||||
expected_oid: str
|
||||
capabilities: frozenset[str]
|
||||
denied: frozenset[str]
|
||||
|
||||
def require(self, capability: str) -> None:
|
||||
if capability in self.denied or capability not in self.capabilities:
|
||||
raise MachineBrokerError("CAPABILITY_DENIED")
|
||||
if self.branch in {"main", "master"}:
|
||||
raise MachineBrokerError("PROTECTED_BRANCH_DENIED")
|
||||
|
||||
|
||||
class GiteaMachineBroker:
|
||||
def __init__(self, transport: GiteaTransport, credential: MachineCredentialProjection, receipt: CampaignCapabilityReceipt) -> None:
|
||||
self.transport = transport
|
||||
self.credential = credential
|
||||
self.receipt = receipt
|
||||
|
||||
def push_branch(self, new_oid: str) -> str:
|
||||
self.receipt.require("exact_task_branch_push")
|
||||
return self.transport.compare_and_swap_branch(self.receipt.branch, self.receipt.expected_oid, new_oid)
|
||||
|
||||
def create_pr(self, title: str, body: str) -> Mapping[str, Any]:
|
||||
self.receipt.require("pr_create_update")
|
||||
return self.transport.create_or_update_pr(self.receipt.branch, "main", title, body)
|
||||
|
||||
def status(self, state: str, description: str) -> Mapping[str, Any]:
|
||||
self.receipt.require("status_publish")
|
||||
return self.transport.publish_status(self.receipt.branch, state, description)
|
||||
|
||||
def merge(self, pr_number: int) -> Mapping[str, Any]:
|
||||
self.receipt.require("protected_merge")
|
||||
return self.transport.protected_merge(pr_number)
|
||||
|
||||
def readback(self) -> str | None:
|
||||
self.receipt.require("remote_readback")
|
||||
return self.transport.read_branch(self.receipt.branch)
|
||||
|
||||
|
||||
def reconcile_exact_branch(authority_oid: str, gitea_oid: str | None, expected_oid: str) -> str:
|
||||
"""Classify exact ref state without force-pushing or guessing."""
|
||||
if authority_oid != expected_oid:
|
||||
return "AUTHORITY_EXPECTATION_MISMATCH"
|
||||
if gitea_oid is None:
|
||||
return "PUSH_REQUIRED"
|
||||
if gitea_oid == expected_oid:
|
||||
return "ALREADY_IN_SYNC"
|
||||
return "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Durable GitOps job controller boundary.
|
||||
|
||||
All filesystem and Git side effects belong to a supervised worker. The
|
||||
controller only validates a binding, persists a job, and advances its durable
|
||||
state. This is the compatibility surface used by write APIs to eliminate
|
||||
ambiguous post-502 retries.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
from typing import Any, Mapping
|
||||
|
||||
|
||||
class JobError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class BindingConflict(JobError):
|
||||
pass
|
||||
|
||||
|
||||
class JobState(str, Enum):
|
||||
REQUESTED = "REQUESTED"
|
||||
AUTHORIZED = "AUTHORIZED"
|
||||
LEASED = "LEASED"
|
||||
WORKTREE_READY = "WORKTREE_READY"
|
||||
EXECUTING = "EXECUTING"
|
||||
LOCAL_VERIFIED = "LOCAL_VERIFIED"
|
||||
REMOTE_SYNCING = "REMOTE_SYNCING"
|
||||
REMOTE_VERIFIED = "REMOTE_VERIFIED"
|
||||
COMPLETED = "COMPLETED"
|
||||
RECOVERABLE_FAILURE = "RECOVERABLE_FAILURE"
|
||||
RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED"
|
||||
MANUAL_OWNER_REVIEW_REQUIRED = "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
|
||||
|
||||
_ORDER = {
|
||||
JobState.REQUESTED: 0,
|
||||
JobState.AUTHORIZED: 1,
|
||||
JobState.LEASED: 2,
|
||||
JobState.WORKTREE_READY: 3,
|
||||
JobState.EXECUTING: 4,
|
||||
JobState.LOCAL_VERIFIED: 5,
|
||||
JobState.REMOTE_SYNCING: 6,
|
||||
JobState.REMOTE_VERIFIED: 7,
|
||||
JobState.COMPLETED: 8,
|
||||
}
|
||||
_RECOVERABLE = {JobState.RECOVERABLE_FAILURE, JobState.RECONCILIATION_REQUIRED}
|
||||
|
||||
|
||||
def canonical_json(value: Any) -> bytes:
|
||||
return (
|
||||
json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n"
|
||||
).encode()
|
||||
|
||||
|
||||
def sha256_json(value: Any) -> str:
|
||||
return hashlib.sha256(canonical_json(value)).hexdigest()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class JobReceipt:
|
||||
job_id: str
|
||||
idempotency_key: str
|
||||
state: JobState
|
||||
side_effect_state: str
|
||||
http_status: int
|
||||
receipt: Mapping[str, Any] | None = None
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"job_id": self.job_id,
|
||||
"idempotency_key": self.idempotency_key,
|
||||
"state": self.state.value,
|
||||
"side_effect_state": self.side_effect_state,
|
||||
"http_status": self.http_status,
|
||||
"receipt": dict(self.receipt) if self.receipt else None,
|
||||
}
|
||||
|
||||
|
||||
class JobStore:
|
||||
def __init__(self, root: str | os.PathLike[str]) -> None:
|
||||
self.root = Path(root)
|
||||
if not self.root.is_absolute() or self.root.is_symlink():
|
||||
raise JobError("JOB_ROOT_INVALID")
|
||||
self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
os.chmod(self.root, 0o700)
|
||||
for name in ("jobs", "idempotency"):
|
||||
child = self.root / name
|
||||
child.mkdir(mode=0o700, exist_ok=True)
|
||||
if child.is_symlink():
|
||||
raise JobError("JOB_ROOT_SYMLINK")
|
||||
|
||||
def _write(self, path: Path, value: Mapping[str, Any]) -> None:
|
||||
if path.is_symlink():
|
||||
raise JobError("JOB_PATH_SYMLINK")
|
||||
fd, temporary = tempfile.mkstemp(prefix=".job-", dir=path.parent)
|
||||
try:
|
||||
os.fchmod(fd, 0o600)
|
||||
with os.fdopen(fd, "wb") as stream:
|
||||
stream.write(canonical_json(value))
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, path)
|
||||
os.chmod(path, 0o600)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
def read(self, job_id: str) -> dict[str, Any]:
|
||||
path = self.root / "jobs" / job_id / "state.json"
|
||||
if path.is_symlink() or not path.is_file():
|
||||
raise JobError("JOB_NOT_FOUND")
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict):
|
||||
raise JobError("JOB_STATE_INVALID")
|
||||
return value
|
||||
|
||||
def _persist(self, job_id: str, value: Mapping[str, Any]) -> None:
|
||||
directory = self.root / "jobs" / job_id
|
||||
directory.mkdir(mode=0o700, exist_ok=True)
|
||||
for name in ("receipts", "logs-digest"):
|
||||
(directory / name).mkdir(mode=0o700, exist_ok=True)
|
||||
self._write(directory / "request.json", value.get("request", {}))
|
||||
self._write(directory / "state.json", value)
|
||||
|
||||
|
||||
class GitOpsJobController:
|
||||
def __init__(self, store: JobStore) -> None:
|
||||
self.store = store
|
||||
|
||||
@staticmethod
|
||||
def _binding(request: Mapping[str, Any]) -> dict[str, Any]:
|
||||
required = (
|
||||
"campaign_id",
|
||||
"repository",
|
||||
"operation",
|
||||
"branch",
|
||||
"expected_branch_oid",
|
||||
"payload_digest",
|
||||
)
|
||||
if any(not isinstance(request.get(name), str) or not request[name].strip() for name in required):
|
||||
raise JobError("JOB_BINDING_INVALID")
|
||||
oid = request["expected_branch_oid"]
|
||||
if len(oid) != 40 or any(ch not in "0123456789abcdef" for ch in oid):
|
||||
raise JobError("EXPECTED_BRANCH_OID_INVALID")
|
||||
digest = request["payload_digest"]
|
||||
if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest):
|
||||
raise JobError("PAYLOAD_DIGEST_INVALID")
|
||||
if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
|
||||
raise JobError("PROTECTED_BRANCH_REJECTED")
|
||||
return {name: request[name] for name in required}
|
||||
|
||||
def submit(self, request: Mapping[str, Any]) -> JobReceipt:
|
||||
binding = self._binding(request)
|
||||
key = sha256_json(binding)
|
||||
index = self.store.root / "idempotency" / f"{key}.json"
|
||||
if index.is_file() and not index.is_symlink():
|
||||
existing = json.loads(index.read_text(encoding="utf-8"))
|
||||
if existing.get("binding") != binding:
|
||||
raise BindingConflict("BINDING_CONFLICT")
|
||||
job_id = existing["job_id"]
|
||||
state = self.store.read(job_id)
|
||||
return self._receipt(state, key, job_id, duplicate=True)
|
||||
job_id = "job-" + key[:32]
|
||||
# Keep the durable request free of arbitrary caller payloads. Workers
|
||||
# consume an approved payload reference or digest-bound artifact; raw
|
||||
# content must never be copied into job state or logs.
|
||||
durable_request = {name: request[name] for name in binding}
|
||||
state = {
|
||||
"schema": "aios.gitops-job-state.v1",
|
||||
"job_id": job_id,
|
||||
"idempotency_key": key,
|
||||
"binding": binding,
|
||||
"request": durable_request,
|
||||
"state": JobState.REQUESTED.value,
|
||||
"revision": 0,
|
||||
"side_effect_state": "NOT_STARTED",
|
||||
"receipt": None,
|
||||
}
|
||||
self.store._persist(job_id, state)
|
||||
self.store._write(index, {"job_id": job_id, "binding": binding})
|
||||
return self._receipt(state, key, job_id, duplicate=False)
|
||||
|
||||
def _receipt(self, state: Mapping[str, Any], key: str, job_id: str, *, duplicate: bool) -> JobReceipt:
|
||||
current = JobState(state["state"])
|
||||
if current == JobState.COMPLETED:
|
||||
status, side = 200, "COMPLETED"
|
||||
elif current in _RECOVERABLE:
|
||||
status, side = 202, "RESUME_SAME_JOB"
|
||||
else:
|
||||
status, side = 202, "QUEUED" if not duplicate else "SAME_JOB"
|
||||
return JobReceipt(job_id, key, current, side, status, state.get("receipt"))
|
||||
|
||||
def advance(
|
||||
self,
|
||||
job_id: str,
|
||||
state: JobState,
|
||||
*,
|
||||
side_effect_state: str,
|
||||
receipt: Mapping[str, Any] | None = None,
|
||||
) -> JobReceipt:
|
||||
current = self.store.read(job_id)
|
||||
previous = JobState(current["state"])
|
||||
if state not in _RECOVERABLE and state != JobState.MANUAL_OWNER_REVIEW_REQUIRED:
|
||||
if previous in _RECOVERABLE:
|
||||
raise JobError("RECOVERABLE_JOB_REQUIRES_RECONCILIATION")
|
||||
if _ORDER.get(state, -1) < _ORDER.get(previous, -1):
|
||||
raise JobError("JOB_STATE_REGRESSION")
|
||||
current["state"] = state.value
|
||||
current["revision"] = int(current["revision"]) + 1
|
||||
current["side_effect_state"] = side_effect_state
|
||||
if receipt is not None:
|
||||
current["receipt"] = dict(receipt)
|
||||
self.store._persist(job_id, current)
|
||||
return self._receipt(current, current["idempotency_key"], job_id, duplicate=True)
|
||||
|
||||
def resume(self, job_id: str) -> JobReceipt:
|
||||
current = self.store.read(job_id)
|
||||
state = JobState(current["state"])
|
||||
if state not in _RECOVERABLE:
|
||||
return self._receipt(current, current["idempotency_key"], job_id, duplicate=True)
|
||||
return self.advance(job_id, JobState.REQUESTED, side_effect_state="RECONCILIATION_RESUME")
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
"""Stateless supervised worker shell for durable GitOps jobs.
|
||||
|
||||
The worker is intentionally inert while the global write gate is false. A
|
||||
future machine-broker-backed executor can resume jobs from the durable store;
|
||||
this shell exists so socket activation has a stable supervised target and
|
||||
cannot silently turn a disabled write plane into direct repository writes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import time
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--job-root", required=True, type=Path)
|
||||
args = parser.parse_args()
|
||||
if os.getenv("AIOS_GITOPS_WRITES_ENABLED", "false").lower() not in {"1", "true", "yes", "on"}:
|
||||
# The campaign deliberately keeps the global gate false until the
|
||||
# machine credential, broker, and release gates are independently live.
|
||||
while True:
|
||||
signal.pause()
|
||||
args.job_root.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
while True:
|
||||
time.sleep(5)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -951,3 +951,51 @@ class AuthorizedTaskWriterProvisioner:
|
||||
return {**public_result, "prestart": prestart}
|
||||
finally:
|
||||
shutil.rmtree(temporary_root, ignore_errors=True)
|
||||
|
||||
|
||||
# v2 continuation validation is intentionally separate from the legacy Owner
|
||||
# bootstrap path above. It consumes a recovered campaign authorization and
|
||||
# never creates a branch or asks the Owner to authorize again.
|
||||
from aios_continuation_lease import WriterLeaseRequestV2
|
||||
|
||||
|
||||
def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict[str, Any]:
|
||||
"""Validate a CONTINUE_EXISTING request against the live exact branch head."""
|
||||
required = {
|
||||
"schema",
|
||||
"task_id",
|
||||
"repository",
|
||||
"branch_mode",
|
||||
"base_oid",
|
||||
"expected_branch_oid",
|
||||
"target_branch",
|
||||
"generation",
|
||||
"lease_seconds",
|
||||
}
|
||||
if set(raw) != required or raw.get("schema") != WriterLeaseRequestV2.schema:
|
||||
raise ProvisioningError("LEASE_REQUEST_V2_FIELDS_REJECTED", 409)
|
||||
request = WriterLeaseRequestV2(
|
||||
task_id=str(raw["task_id"]),
|
||||
repository=str(raw["repository"]),
|
||||
branch_mode=str(raw["branch_mode"]),
|
||||
base_oid=str(raw["base_oid"]),
|
||||
expected_branch_oid=str(raw["expected_branch_oid"]),
|
||||
target_branch=str(raw["target_branch"]),
|
||||
generation=raw["generation"],
|
||||
lease_seconds=raw["lease_seconds"],
|
||||
)
|
||||
request.validate(probe)
|
||||
return {
|
||||
"schema": WriterLeaseRequestV2.schema,
|
||||
"task_id": request.task_id,
|
||||
"repository": request.repository,
|
||||
"branch_mode": request.branch_mode,
|
||||
"base_oid": request.base_oid,
|
||||
"expected_branch_oid": request.expected_branch_oid,
|
||||
"target_branch": request.target_branch,
|
||||
"generation": request.generation,
|
||||
"lease_seconds": request.lease_seconds,
|
||||
"create_target_branch": False,
|
||||
"worktree_mode": "detached_at_expected_head",
|
||||
"cas_ref_update_required": True,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
[Unit]
|
||||
Description=AIOS durable GitOps writer socket
|
||||
|
||||
[Socket]
|
||||
ListenStream=/run/aios-gitops/writer.sock
|
||||
SocketUser=root
|
||||
SocketGroup=gitops
|
||||
SocketMode=0660
|
||||
DirectoryMode=0750
|
||||
RemoveOnStop=yes
|
||||
|
||||
[Install]
|
||||
WantedBy=sockets.target
|
||||
@@ -0,0 +1,24 @@
|
||||
[Unit]
|
||||
Description=AIOS durable GitOps writer worker (%i)
|
||||
Requires=aios-gitops-writer.socket
|
||||
After=aios-gitops-writer.socket
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=root
|
||||
Group=root
|
||||
UMask=0077
|
||||
Environment=AIOS_GITOPS_WRITES_ENABLED=false
|
||||
Environment=AIOS_GITOPS_JOB_ROOT=/var/lib/aios-gitops/jobs
|
||||
Environment=PYTHONDONTWRITEBYTECODE=1
|
||||
ExecStart=/usr/bin/python3 -B /opt/aios-gitops/active-runtime/app/aios_gitops_writer_worker.py --job-root ${AIOS_GITOPS_JOB_ROOT}
|
||||
Restart=on-failure
|
||||
RestartSec=2
|
||||
NoNewPrivileges=no
|
||||
PrivateTmp=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
ReadWritePaths=/var/lib/aios-gitops/jobs /var/lib/aios-gitops/worktrees
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,27 @@
|
||||
# Durable GitOps Job Controller
|
||||
|
||||
The control-plane write boundary is a durable Job, not a synchronous checkout
|
||||
or commit request. The request is persisted under
|
||||
`/var/lib/aios-gitops/jobs/<job_id>/` with `request.json`, `state.json`,
|
||||
`receipts/`, and `logs-digest/`.
|
||||
|
||||
The controller owns only binding validation, idempotency, state transitions,
|
||||
and recovery classification. A supervised worker owns an isolated detached
|
||||
worktree and all Git side effects. `/api/file` and `/api/commit-file` adapters
|
||||
must return `job_id` and `side_effect_state`; they must never perform
|
||||
checkout, file write, `git add`, or `git commit` inline.
|
||||
|
||||
The idempotency key is the SHA-256 of the canonical tuple:
|
||||
|
||||
`campaign_id`, `repository`, `operation`, `branch`, `expected_branch_oid`, and
|
||||
`payload_digest`.
|
||||
|
||||
The writer lease is `CONTINUE_EXISTING` for the recovered campaign. It binds
|
||||
to lineage anchor `69b45e81dcd2c3707c367003b08c55fd361faae8` and exact initial
|
||||
head `bba675fab3b7e0c5fba7066720252b3c91b9f20d`. Every ref update is delegated
|
||||
to a non-Owner machine broker using compare-and-swap; a stale head is a
|
||||
`BRANCH_LEASE_STALE` condition and never a force push.
|
||||
|
||||
Production remains outside this worker contract. `AIOS_GITOPS_WRITES_ENABLED`
|
||||
is false in the supplied unit template until all external Gitea and release
|
||||
gates are independently proven.
|
||||
@@ -0,0 +1,156 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from aios_continuation_lease import (
|
||||
CAMPAIGN_HEAD,
|
||||
LINEAGE_ANCHOR,
|
||||
ContinuationError,
|
||||
ContinuationLeaseManager,
|
||||
WriterLeaseRequestV2,
|
||||
)
|
||||
from aios_gitops_job_controller import JobState, JobStore, GitOpsJobController, JobError
|
||||
from aios_gitea_machine_broker import MachineBrokerError, MachineCredentialProjection, reconcile_exact_branch
|
||||
from authorized_task_writer_provisioning import validate_writer_lease_request_v2
|
||||
|
||||
|
||||
class Probe:
|
||||
def __init__(self):
|
||||
self.head = CAMPAIGN_HEAD
|
||||
|
||||
def branch_exists(self, branch):
|
||||
return branch == "agent/autonomous-release-campaign-v1-20260807"
|
||||
|
||||
def branch_head(self, branch):
|
||||
return self.head
|
||||
|
||||
def is_ancestor(self, older, newer):
|
||||
return older == LINEAGE_ANCHOR and newer == CAMPAIGN_HEAD
|
||||
|
||||
|
||||
class CAS:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
self.stale = False
|
||||
|
||||
def compare_and_swap_branch(self, branch, expected, new):
|
||||
if self.stale:
|
||||
raise RuntimeError("stale")
|
||||
self.calls.append((branch, expected, new))
|
||||
return new
|
||||
|
||||
|
||||
def request(generation=1):
|
||||
return WriterLeaseRequestV2(
|
||||
task_id="AIOS_AUTONOMOUS_RELEASE_CAMPAIGN_V1",
|
||||
repository="aios/aios",
|
||||
branch_mode="CONTINUE_EXISTING",
|
||||
base_oid=LINEAGE_ANCHOR,
|
||||
expected_branch_oid=CAMPAIGN_HEAD,
|
||||
target_branch="agent/autonomous-release-campaign-v1-20260807",
|
||||
generation=generation,
|
||||
)
|
||||
|
||||
|
||||
def test_continue_existing_accepts_exact_head_and_lineage():
|
||||
manager = ContinuationLeaseManager(
|
||||
Probe(), CAS(), now=lambda tz: datetime(2026, 8, 8, tzinfo=timezone.utc)
|
||||
)
|
||||
lease = manager.open(request())
|
||||
assert lease.current_head_oid == CAMPAIGN_HEAD
|
||||
assert lease.multi_commit and lease.append_only and lease.recoverable
|
||||
|
||||
|
||||
def test_continue_existing_rejects_branch_head_conflict():
|
||||
probe = Probe()
|
||||
probe.head = "c" * 40
|
||||
with pytest.raises(ContinuationError, match="EXISTING_BRANCH_OID_CONFLICT"):
|
||||
request().validate(probe)
|
||||
|
||||
|
||||
def test_provisioning_contract_v2_never_creates_existing_branch():
|
||||
raw = {
|
||||
"schema": "aios.task-writer-lease-request.v2",
|
||||
"task_id": request().task_id,
|
||||
"repository": request().repository,
|
||||
"branch_mode": request().branch_mode,
|
||||
"base_oid": request().base_oid,
|
||||
"expected_branch_oid": request().expected_branch_oid,
|
||||
"target_branch": request().target_branch,
|
||||
"generation": 1,
|
||||
"lease_seconds": 900,
|
||||
}
|
||||
result = validate_writer_lease_request_v2(raw, Probe())
|
||||
assert result["create_target_branch"] is False
|
||||
assert result["cas_ref_update_required"] is True
|
||||
|
||||
|
||||
def test_append_uses_compare_and_swap_and_allows_multiple_commits():
|
||||
cas = CAS()
|
||||
manager = ContinuationLeaseManager(Probe(), cas)
|
||||
lease = manager.open(request())
|
||||
next_lease = manager.append_commit(lease, "c" * 40)
|
||||
manager.append_commit(next_lease, "d" * 40)
|
||||
assert cas.calls == [
|
||||
(request().target_branch, CAMPAIGN_HEAD, "c" * 40),
|
||||
(request().target_branch, "c" * 40, "d" * 40),
|
||||
]
|
||||
|
||||
|
||||
def test_stale_cas_fails_closed():
|
||||
cas = CAS()
|
||||
cas.stale = True
|
||||
manager = ContinuationLeaseManager(Probe(), cas)
|
||||
with pytest.raises(ContinuationError, match="BRANCH_LEASE_STALE"):
|
||||
manager.append_commit(manager.open(request()), "c" * 40)
|
||||
|
||||
|
||||
def test_job_idempotency_and_recovery():
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
controller = GitOpsJobController(JobStore(temp))
|
||||
req = {
|
||||
"campaign_id": "campaign",
|
||||
"repository": "aios/aios",
|
||||
"operation": "commit-file",
|
||||
"branch": request().target_branch,
|
||||
"expected_branch_oid": CAMPAIGN_HEAD,
|
||||
"payload_digest": "b" * 64,
|
||||
"payload": {"files": ["active-runtime/app/x.py"]},
|
||||
}
|
||||
first = controller.submit(req)
|
||||
duplicate = controller.submit(req)
|
||||
assert first.job_id == duplicate.job_id
|
||||
assert duplicate.side_effect_state == "SAME_JOB"
|
||||
controller.advance(first.job_id, JobState.RECOVERABLE_FAILURE, side_effect_state="WORKER_CRASH")
|
||||
resumed = controller.submit(req)
|
||||
assert resumed.job_id == first.job_id and resumed.side_effect_state == "RESUME_SAME_JOB"
|
||||
|
||||
|
||||
def test_job_state_cannot_regress():
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
controller = GitOpsJobController(JobStore(temp))
|
||||
req = {
|
||||
"campaign_id": "campaign",
|
||||
"repository": "aios/aios",
|
||||
"operation": "commit-file",
|
||||
"branch": request().target_branch,
|
||||
"expected_branch_oid": CAMPAIGN_HEAD,
|
||||
"payload_digest": "b" * 64,
|
||||
}
|
||||
job = controller.submit(req)
|
||||
controller.advance(job.job_id, JobState.AUTHORIZED, side_effect_state="AUTHORIZED")
|
||||
with pytest.raises(JobError):
|
||||
controller.advance(job.job_id, JobState.REQUESTED, side_effect_state="BAD")
|
||||
|
||||
|
||||
def test_machine_broker_rejects_owner_identity_and_plaintext_reference():
|
||||
with pytest.raises(MachineBrokerError):
|
||||
MachineCredentialProjection("owner-token-value", "aiman")
|
||||
|
||||
|
||||
def test_ref_reconciler_is_fail_closed():
|
||||
assert reconcile_exact_branch("a" * 40, None, "a" * 40) == "PUSH_REQUIRED"
|
||||
assert reconcile_exact_branch("a" * 40, "b" * 40, "a" * 40) == "MANUAL_OWNER_REVIEW_REQUIRED"
|
||||
Reference in New Issue
Block a user