718 lines
31 KiB
Python
718 lines
31 KiB
Python
"""Governed continuation executor for exact existing task branches.
|
|
|
|
The executor composes the existing continuation lease and durable Job store.
|
|
All Git and Gitea side effects live behind injected narrow interfaces so the
|
|
state machine can prove Authority, Git-network, API-branch, and commit-object
|
|
agreement before it reports completion.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import subprocess
|
|
from typing import Callable, Mapping, Protocol
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote, urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
from aios_authority_git_executor import AuthorityGitExecutor
|
|
from aios_continuation_lease import WriterLeaseRequestV2
|
|
from aios_gitops_job_controller import GitOpsJobController, JobState
|
|
from aios_gitops_reconciliation_core import ReconciliationAction, RemoteObservation, decide_reconciliation
|
|
from authorized_task_writer_provisioning import validate_writer_lease_request_v2
|
|
|
|
|
|
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
|
SECRET_PATTERNS = (
|
|
re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"),
|
|
re.compile(rb"(?i)(?:token|password|secret|api[_-]?key)\s*[:=]\s*[\"'][^\"'\s]{12,}[\"']"),
|
|
re.compile(rb"-----BEGIN " + rb"(?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
|
|
)
|
|
|
|
|
|
class ContinuationExecutionError(RuntimeError):
|
|
"""Stable fail-closed continuation execution error."""
|
|
|
|
|
|
class DetachedWorkspace(Protocol):
|
|
def is_detached(self) -> bool: ...
|
|
def head_oid(self) -> str: ...
|
|
def commit_files(self, files: Mapping[str, bytes], message: str) -> str: ...
|
|
|
|
|
|
class AuthorityPort(Protocol):
|
|
def branch_exists(self, branch: str) -> bool: ...
|
|
def branch_head(self, branch: str) -> str: ...
|
|
def is_ancestor(self, older: str, newer: str) -> bool: ...
|
|
def create_task_branch_cas(self, branch: str, expected_base: str, new: str) -> str: ...
|
|
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...
|
|
|
|
|
|
class RemotePort(Protocol):
|
|
def repository_identity(self) -> str: ...
|
|
def push_new_exact(self, branch: str, new: str) -> None: ...
|
|
def push_exact(self, branch: str, expected: str, new: str) -> None: ...
|
|
def read_git_ref(self, branch: str) -> str | None: ...
|
|
def read_api_branch(self, branch: str) -> str | None: ...
|
|
def commit_visible(self, oid: str) -> bool: ...
|
|
|
|
|
|
class CredentialProjection(Protocol):
|
|
def project(self) -> None: ...
|
|
def revoke(self) -> None: ...
|
|
|
|
|
|
def _run(argv: list[str], *, cwd: Path | None = None, environment: Mapping[str, str] | None = None) -> str:
|
|
result = subprocess.run(
|
|
argv,
|
|
cwd=cwd,
|
|
env=None if environment is None else dict(environment),
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise ContinuationExecutionError("SUBPROCESS_FAILED")
|
|
return result.stdout.strip()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AuthorityGitAdapter:
|
|
"""Branch-name adapter over the existing narrow AuthorityGitExecutor."""
|
|
|
|
executor: AuthorityGitExecutor
|
|
|
|
@staticmethod
|
|
def _ref(branch: str) -> str:
|
|
return f"refs/heads/{branch}"
|
|
|
|
def branch_exists(self, branch: str) -> bool:
|
|
return self.executor.read_ref(self._ref(branch)) is not None
|
|
|
|
def branch_head(self, branch: str) -> str:
|
|
value = self.executor.read_ref(self._ref(branch))
|
|
if value is None:
|
|
raise ContinuationExecutionError("EXISTING_BRANCH_MISSING")
|
|
return value
|
|
|
|
def is_ancestor(self, older: str, newer: str) -> bool:
|
|
return self.executor.ancestry_check(older, newer)
|
|
|
|
def create_task_branch_cas(self, branch: str, expected_base: str, new: str) -> str:
|
|
return self.executor.create_task_branch_CAS(self._ref(branch), expected_base, new)
|
|
|
|
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str:
|
|
return self.executor.advance_task_branch_CAS(self._ref(branch), expected, new)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DetachedGitWorkspace:
|
|
path: Path
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.path.is_absolute() or self.path.is_symlink() or not self.path.is_dir():
|
|
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
|
|
top = Path(_run(["git", "rev-parse", "--show-toplevel"], cwd=self.path)).resolve()
|
|
if top != self.path.resolve():
|
|
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
|
|
|
|
def is_detached(self) -> bool:
|
|
result = subprocess.run(
|
|
["git", "symbolic-ref", "-q", "HEAD"],
|
|
cwd=self.path,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
return result.returncode == 1
|
|
|
|
def head_oid(self) -> str:
|
|
return _run(["git", "rev-parse", "HEAD"], cwd=self.path)
|
|
|
|
def _dirty_paths(self) -> set[str]:
|
|
command = subprocess.run(
|
|
["git", "status", "--porcelain=v1", "--untracked-files=all"],
|
|
cwd=self.path,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if command.returncode != 0:
|
|
raise ContinuationExecutionError("SUBPROCESS_FAILED")
|
|
output = command.stdout
|
|
result: set[str] = set()
|
|
for line in output.splitlines():
|
|
path = line[3:]
|
|
if " -> " in path:
|
|
path = path.split(" -> ", 1)[1]
|
|
result.add(path)
|
|
return result
|
|
|
|
def materialize_files(self, files: Mapping[str, bytes]) -> None:
|
|
dirty = self._dirty_paths()
|
|
expected_paths = set(files)
|
|
if dirty and not dirty.issubset(expected_paths):
|
|
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
|
|
for relative, content in files.items():
|
|
target = (self.path / relative).resolve()
|
|
try:
|
|
target.relative_to(self.path.resolve())
|
|
except ValueError as exc:
|
|
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
|
|
if target.is_symlink():
|
|
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(content)
|
|
|
|
def commit_files(self, files: Mapping[str, bytes], message: str) -> str:
|
|
expected_paths = set(files)
|
|
if self._dirty_paths() != expected_paths:
|
|
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
|
|
for relative, content in files.items():
|
|
target = (self.path / relative).resolve()
|
|
try:
|
|
target.relative_to(self.path.resolve())
|
|
except ValueError as exc:
|
|
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
|
|
if not target.is_file() or target.is_symlink() or target.read_bytes() != content:
|
|
raise ContinuationExecutionError("AUTHORING_PAYLOAD_MISMATCH")
|
|
_run(["git", "add", "--", *sorted(expected_paths)], cwd=self.path)
|
|
staged = set(_run(["git", "diff", "--cached", "--name-only"], cwd=self.path).splitlines())
|
|
if staged != expected_paths:
|
|
raise ContinuationExecutionError("STAGED_SCOPE_MISMATCH")
|
|
environment = os.environ.copy()
|
|
environment.update(
|
|
{
|
|
"GIT_AUTHOR_NAME": "AIOS Governed Continuation",
|
|
"GIT_AUTHOR_EMAIL": "aios-governed-continuation@localhost",
|
|
"GIT_COMMITTER_NAME": "AIOS Governed Continuation",
|
|
"GIT_COMMITTER_EMAIL": "aios-governed-continuation@localhost",
|
|
}
|
|
)
|
|
_run(["git", "commit", "--no-gpg-sign", "-m", message, "--", *sorted(expected_paths)], cwd=self.path, environment=environment)
|
|
if not self.is_detached():
|
|
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
|
|
return self.head_oid()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ServerSideCredentialProjection:
|
|
ssh_key_path: Path
|
|
api_token_path: Path
|
|
active: bool = False
|
|
|
|
def project(self) -> None:
|
|
for path in (self.ssh_key_path, self.api_token_path):
|
|
if path.is_symlink() or not path.is_file() or path.stat().st_size < 32:
|
|
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
|
|
self.active = True
|
|
|
|
def revoke(self) -> None:
|
|
self.active = False
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GiteaContinuationRemote:
|
|
workspace: Path
|
|
credential: ServerSideCredentialProjection
|
|
remote_name: str = "gitea"
|
|
repository: str = "aios/aios"
|
|
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
|
|
expected_host: str = "g.cnnetmkt.top"
|
|
|
|
@staticmethod
|
|
def validate_remote_url(value: str, expected_host: str = "g.cnnetmkt.top") -> None:
|
|
if value.startswith(("/", "file:")) or value.endswith("/data/aios-gitops/repo.git"):
|
|
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
|
|
if "://" in value:
|
|
parsed = urlsplit(value)
|
|
host, path = parsed.hostname, parsed.path
|
|
else:
|
|
location = value.split("@", 1)[-1]
|
|
host, separator, path = location.partition(":")
|
|
if not separator:
|
|
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
|
|
path = "/" + path
|
|
if host != expected_host or path.rstrip("/") not in {"/aios/aios.git", "/git/aios/aios.git"}:
|
|
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
|
|
|
|
def repository_identity(self) -> str:
|
|
url = _run(["git", "remote", "get-url", self.remote_name], cwd=self.workspace)
|
|
self.validate_remote_url(url, self.expected_host)
|
|
if self.repository != "aios/aios":
|
|
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
|
|
return self.repository
|
|
|
|
def _require_projected(self) -> None:
|
|
if not self.credential.active:
|
|
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
|
|
|
|
def read_git_ref(self, branch: str) -> str | None:
|
|
self._require_projected()
|
|
output = _run(["git", "ls-remote", self.remote_name, f"refs/heads/{branch}"], cwd=self.workspace)
|
|
if not output:
|
|
return None
|
|
oid, ref = output.split(maxsplit=1)
|
|
if ref != f"refs/heads/{branch}" or not OID_RE.fullmatch(oid):
|
|
raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
|
|
return oid
|
|
|
|
def push_new_exact(self, branch: str, new: str) -> None:
|
|
self._require_projected()
|
|
self.repository_identity()
|
|
if self.read_git_ref(branch) is not None:
|
|
raise ContinuationExecutionError("GITEA_TARGET_BRANCH_ALREADY_EXISTS")
|
|
_run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
|
|
if self.read_git_ref(branch) != new:
|
|
raise ContinuationExecutionError("GITEA_CREATE_READBACK_MISMATCH")
|
|
|
|
def push_exact(self, branch: str, expected: str, new: str) -> None:
|
|
self._require_projected()
|
|
self.repository_identity()
|
|
if self.read_git_ref(branch) != expected:
|
|
raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
|
|
_run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
|
|
|
|
def _api(self, path: str) -> object:
|
|
self._require_projected()
|
|
token = self.credential.api_token_path.read_text(encoding="utf-8").strip()
|
|
request = Request(
|
|
self.api_base.rstrip("/") + path,
|
|
headers={"Accept": "application/json", "Authorization": f"token {token}"},
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=20) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except HTTPError as exc:
|
|
if exc.code == 404:
|
|
return None
|
|
raise ContinuationExecutionError(f"GITEA_API_HTTP_{exc.code}") from exc
|
|
except (URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
|
|
raise ContinuationExecutionError("GITEA_API_UNAVAILABLE") from exc
|
|
|
|
def read_api_branch(self, branch: str) -> str | None:
|
|
owner, name = self.repository.split("/", 1)
|
|
page = 1
|
|
while page <= 100:
|
|
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/branches?page={page}&limit=50")
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, list):
|
|
raise ContinuationExecutionError("GITEA_API_BRANCH_COLLECTION_INVALID")
|
|
for item in value:
|
|
if not isinstance(item, dict) or item.get("name") != branch:
|
|
continue
|
|
oid = item.get("commit", {}).get("id")
|
|
if not isinstance(oid, str) or not OID_RE.fullmatch(oid):
|
|
raise ContinuationExecutionError("GITEA_API_BRANCH_INVALID")
|
|
return oid
|
|
if len(value) < 50:
|
|
return None
|
|
page += 1
|
|
raise ContinuationExecutionError("GITEA_API_BRANCH_PAGINATION_LIMIT")
|
|
|
|
def commit_visible(self, oid: str) -> bool:
|
|
owner, name = self.repository.split("/", 1)
|
|
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/git/commits/{oid}")
|
|
if not isinstance(value, dict):
|
|
return False
|
|
observed = value.get("sha") or value.get("id")
|
|
return observed == oid
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ContinuationPermissions:
|
|
create_target_branch: bool = False
|
|
force_push: bool = False
|
|
tag_write: bool = False
|
|
baseline_write: bool = False
|
|
production_write: bool = False
|
|
main_write: bool = False
|
|
|
|
def validate(self) -> None:
|
|
if any(asdict(self).values()):
|
|
raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ContinuationRuntimeBinding:
|
|
request: WriterLeaseRequestV2
|
|
job_id: str
|
|
workspace_path: Path
|
|
lease_not_before: datetime
|
|
lease_expires_at: datetime
|
|
original_receipt_valid: bool
|
|
original_grant_consumed: bool
|
|
permissions: ContinuationPermissions
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ContinuationCommit:
|
|
idempotency_key: str
|
|
files: Mapping[str, bytes]
|
|
message: str
|
|
|
|
def payload(self) -> bytes:
|
|
if not self.idempotency_key or not self.message.strip() or not self.files:
|
|
raise ContinuationExecutionError("COMMIT_REQUEST_INVALID")
|
|
normalized: dict[str, str] = {}
|
|
for path, content in self.files.items():
|
|
pure = PurePosixPath(path)
|
|
if (
|
|
not isinstance(path, str)
|
|
or pure.is_absolute()
|
|
or ".." in pure.parts
|
|
or path.startswith((".git/", "baseline/", "production/"))
|
|
or not isinstance(content, bytes)
|
|
):
|
|
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
|
|
if any(pattern.search(content) for pattern in SECRET_PATTERNS):
|
|
raise ContinuationExecutionError("SECRET_LIKE_PAYLOAD_REJECTED")
|
|
normalized[path] = base64.b64encode(content).decode("ascii")
|
|
return json.dumps(
|
|
{"files_base64": normalized, "message": self.message},
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
|
|
|
|
class ContinuationExecutor:
|
|
def __init__(
|
|
self,
|
|
controller: GitOpsJobController,
|
|
authority: AuthorityPort,
|
|
workspace: DetachedWorkspace,
|
|
remote: RemotePort,
|
|
credential: CredentialProjection,
|
|
*,
|
|
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
|
|
) -> None:
|
|
self.controller = controller
|
|
self.authority = authority
|
|
self.workspace = workspace
|
|
self.remote = remote
|
|
self.credential = credential
|
|
self.now = now
|
|
self._observed_generations: dict[str, int] = {}
|
|
|
|
def observe_generation(self, branch: str, generation: int) -> None:
|
|
self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))
|
|
|
|
def _max_generation(self, branch: str, *, exclude_job_id: str | None = None) -> int:
|
|
observed = self._observed_generations.get(branch, 0)
|
|
with self.controller.store._connect() as connection:
|
|
if exclude_job_id is None:
|
|
row = connection.execute(
|
|
"SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
|
|
(branch,),
|
|
).fetchone()
|
|
else:
|
|
row = connection.execute(
|
|
"SELECT MAX(generation) AS generation FROM branch_leases "
|
|
"WHERE branch=? AND job_id!=?",
|
|
(branch, exclude_job_id),
|
|
).fetchone()
|
|
durable = int(row["generation"] or 0)
|
|
return max(observed, durable)
|
|
|
|
@staticmethod
|
|
def _request_mapping(request: WriterLeaseRequestV2) -> dict[str, object]:
|
|
return {
|
|
"schema": WriterLeaseRequestV2.schema,
|
|
"task_id": request.task_id,
|
|
"job_id": request.job_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,
|
|
}
|
|
|
|
def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
|
|
request = binding.request
|
|
request_mapping = self._request_mapping(request)
|
|
normalized = validate_writer_lease_request_v2(request_mapping, self.authority)
|
|
expected_normalized = {
|
|
**request_mapping,
|
|
"create_target_branch": request.branch_mode == "CREATE_NEW",
|
|
"worktree_mode": (
|
|
"detached_at_base_oid"
|
|
if request.branch_mode == "CREATE_NEW"
|
|
else "detached_at_expected_head"
|
|
),
|
|
"cas_ref_update_required": True,
|
|
}
|
|
if normalized != expected_normalized:
|
|
raise ContinuationExecutionError("CONTINUATION_NORMALIZATION_INVALID")
|
|
if not binding.original_receipt_valid or not binding.original_grant_consumed:
|
|
raise ContinuationExecutionError("ORIGINAL_GRANT_LINEAGE_INVALID")
|
|
binding.permissions.validate()
|
|
now = self.now()
|
|
if now < binding.lease_not_before:
|
|
raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
|
|
if now >= binding.lease_expires_at:
|
|
raise ContinuationExecutionError("LEASE_EXPIRED")
|
|
if request.generation <= self._max_generation(request.target_branch, exclude_job_id=job_id):
|
|
raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
|
|
if not self.workspace.is_detached():
|
|
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
|
|
if self.workspace.head_oid() != request.expected_branch_oid:
|
|
raise ContinuationExecutionError("WORKTREE_HEAD_MISMATCH")
|
|
if self.remote.repository_identity() != request.repository:
|
|
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
|
|
|
|
def _job_request(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit, payload: bytes) -> dict[str, object]:
|
|
request = binding.request
|
|
return {
|
|
"idempotency_key": commit.idempotency_key,
|
|
"campaign_id": request.task_id,
|
|
"task_id": request.task_id,
|
|
"repository": request.repository,
|
|
"operation": "CONTINUATION_COMMIT_FILES",
|
|
"branch": request.target_branch,
|
|
"expected_branch_oid": request.expected_branch_oid,
|
|
"base_oid": request.base_oid,
|
|
"branch_mode": request.branch_mode,
|
|
"payload_digest": hashlib.sha256(payload).hexdigest(),
|
|
"lineage_anchor_oid": request.base_oid,
|
|
"generation": request.generation,
|
|
}
|
|
|
|
def execute(
|
|
self,
|
|
binding: ContinuationRuntimeBinding,
|
|
commit: ContinuationCommit,
|
|
*,
|
|
before_complete: Callable[[str], None] | None = None,
|
|
) -> dict[str, object]:
|
|
commit.payload()
|
|
if not binding.job_id or binding.request.job_id != binding.job_id:
|
|
raise ContinuationExecutionError("RUNTIME_LINEAGE_MISSING_JOB_ID")
|
|
current = self.controller.store.read(binding.job_id)
|
|
current_state = JobState(current["state"])
|
|
if current_state == JobState.COMPLETED:
|
|
cycle = self.controller.begin_continuation_cycle(
|
|
binding.job_id,
|
|
idempotency_key=commit.idempotency_key,
|
|
expected_head_oid=binding.request.expected_branch_oid,
|
|
generation=binding.request.generation,
|
|
expected_revision=int(current["revision"]),
|
|
)
|
|
if cycle.get("completed"):
|
|
return self.controller.store._receipt(
|
|
self.controller.store.read(binding.job_id), duplicate=True
|
|
).as_dict()
|
|
current = self.controller.store.read(binding.job_id)
|
|
current_state = JobState(current["state"])
|
|
if current_state == JobState.REMOTE_VERIFIED:
|
|
receipt = json.loads(str(current["receipt_json"] or "{}"))
|
|
commit_oid = str(receipt.get("authority_oid") or "")
|
|
if not OID_RE.fullmatch(commit_oid):
|
|
raise ContinuationExecutionError("GOVERNED_COMMIT_OID_INVALID")
|
|
if before_complete is not None:
|
|
before_complete(commit_oid)
|
|
result = self.controller.advance(
|
|
binding.job_id,
|
|
JobState.COMPLETED,
|
|
side_effect_state="REMOTE_VERIFIED",
|
|
receipt=receipt,
|
|
).as_dict()
|
|
self.controller.complete_continuation_cycle(
|
|
binding.job_id,
|
|
idempotency_key=commit.idempotency_key,
|
|
commit_oid=commit_oid,
|
|
)
|
|
return result
|
|
self._validate_preflight(binding, job_id=binding.job_id)
|
|
request = binding.request
|
|
projected = False
|
|
try:
|
|
self.credential.project()
|
|
projected = True
|
|
row = self.controller.store.read(binding.job_id)
|
|
state = JobState(row["state"])
|
|
if state == JobState.REQUESTED:
|
|
self.controller.advance(
|
|
binding.job_id,
|
|
JobState.AUTHORIZED,
|
|
side_effect_state="AUTHORIZED",
|
|
receipt={"task_id": request.task_id, "original_grant_consumed": True},
|
|
expected_revision=int(row["revision"]),
|
|
)
|
|
row = self.controller.store.read(binding.job_id)
|
|
state = JobState(row["state"])
|
|
if state == JobState.AUTHORIZED:
|
|
self.controller.lease_job(
|
|
binding.job_id,
|
|
lease_id=f"lease-{binding.job_id}",
|
|
lineage_anchor_oid=request.base_oid,
|
|
current_head_oid=request.expected_branch_oid,
|
|
generation=request.generation,
|
|
lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
|
|
expected_revision=int(row["revision"]),
|
|
)
|
|
row = self.controller.store.read(binding.job_id)
|
|
state = JobState(row["state"])
|
|
if state == JobState.LEASED:
|
|
self.controller.advance(binding.job_id, JobState.WORKTREE_READY,
|
|
side_effect_state="WORKTREE_READY", expected_revision=int(row["revision"]))
|
|
row = self.controller.store.read(binding.job_id)
|
|
state = JobState(row["state"])
|
|
if state == JobState.WORKTREE_READY:
|
|
self.controller.advance(binding.job_id, JobState.EXECUTING,
|
|
side_effect_state="IN_PROGRESS", expected_revision=int(row["revision"]))
|
|
elif state != JobState.EXECUTING:
|
|
raise ContinuationExecutionError("JOB_EXECUTION_STATE_INVALID")
|
|
new_oid = self.workspace.commit_files(commit.files, commit.message)
|
|
if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
|
|
raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
|
|
if request.branch_mode == "CREATE_NEW":
|
|
authority_oid = self.authority.create_task_branch_cas(request.target_branch, request.base_oid, new_oid)
|
|
else:
|
|
authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
|
|
if authority_oid != new_oid:
|
|
raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
|
|
self.controller.advance(
|
|
binding.job_id,
|
|
JobState.LOCAL_VERIFIED,
|
|
side_effect_state="LOCAL_VERIFIED",
|
|
receipt={"authority_oid": authority_oid, "force_push": False},
|
|
)
|
|
try:
|
|
self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
|
|
if request.branch_mode == "CREATE_NEW":
|
|
self.remote.push_new_exact(request.target_branch, new_oid)
|
|
else:
|
|
self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
|
|
except Exception:
|
|
return self._remote_pending(binding.job_id, authority_oid, new_oid)
|
|
result = self._finish_remote_verification(
|
|
binding.job_id,
|
|
authority_oid,
|
|
new_oid,
|
|
before_complete=before_complete,
|
|
)
|
|
if str(result.get("state") or "") == JobState.COMPLETED.value:
|
|
self.controller.complete_continuation_cycle(
|
|
binding.job_id,
|
|
idempotency_key=commit.idempotency_key,
|
|
commit_oid=new_oid,
|
|
)
|
|
return result
|
|
finally:
|
|
if projected:
|
|
self.credential.revoke()
|
|
|
|
def _observe_remote(self, branch: str, oid: str) -> tuple[str | None, str | None, bool]:
|
|
return self.remote.read_git_ref(branch), self.remote.read_api_branch(branch), self.remote.commit_visible(oid)
|
|
|
|
def _remote_pending(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
|
|
receipt = {
|
|
"authority_oid": authority_oid,
|
|
"desired_remote_oid": new_oid,
|
|
"remote_verified": False,
|
|
"force_push": False,
|
|
}
|
|
result = self.controller.advance(
|
|
job_id,
|
|
JobState.RECONCILIATION_REQUIRED,
|
|
side_effect_state="REMOTE_PUSH_PENDING",
|
|
receipt=receipt,
|
|
)
|
|
return result.as_dict()
|
|
|
|
def _finish_remote_verification(
|
|
self,
|
|
job_id: str,
|
|
authority_oid: str,
|
|
new_oid: str,
|
|
*,
|
|
before_complete: Callable[[str], None] | None = None,
|
|
) -> dict[str, object]:
|
|
row = self.controller.store.read(job_id)
|
|
branch = row["branch"]
|
|
request = json.loads(row["request_json"])["binding"]
|
|
git_oid, api_oid, visible = self._observe_remote(branch, new_oid)
|
|
decision = decide_reconciliation(RemoteObservation("aios/aios", branch, new_oid, authority_oid, git_oid, api_oid, visible, datetime.now(timezone.utc).isoformat(), request.get("expected_branch_oid")))
|
|
if decision.action is not ReconciliationAction.MARK_REMOTE_VERIFIED:
|
|
return self._remote_pending(job_id, authority_oid, new_oid)
|
|
receipt = {
|
|
"authority_oid": authority_oid,
|
|
"git_remote_oid": git_oid,
|
|
"api_branch_oid": api_oid,
|
|
"commit_visible": True,
|
|
"remote_verified": True,
|
|
"reconciliation_decision": decision.action.value,
|
|
"force_push": False,
|
|
}
|
|
self.controller.advance(
|
|
job_id,
|
|
JobState.REMOTE_VERIFIED,
|
|
side_effect_state="REMOTE_VERIFIED",
|
|
receipt=receipt,
|
|
)
|
|
if before_complete is not None:
|
|
before_complete(new_oid)
|
|
return self.controller.advance(
|
|
job_id,
|
|
JobState.COMPLETED,
|
|
side_effect_state="REMOTE_VERIFIED",
|
|
receipt=receipt,
|
|
).as_dict()
|
|
|
|
def reconcile(self, job_id: str) -> dict[str, object]:
|
|
row = self.controller.store.read(job_id)
|
|
state = JobState(row["state"])
|
|
if state == JobState.COMPLETED:
|
|
return self.controller.store._receipt(row, duplicate=True).as_dict()
|
|
if state != JobState.RECONCILIATION_REQUIRED:
|
|
raise ContinuationExecutionError("JOB_NOT_RECONCILABLE")
|
|
request = json.loads(row["request_json"])["binding"]
|
|
receipt = json.loads(row["receipt_json"])
|
|
new_oid = receipt["desired_remote_oid"]
|
|
authority_oid = self.authority.branch_head(row["branch"])
|
|
try:
|
|
generation = int(request.get("generation") or 0)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ContinuationExecutionError("DURABLE_BINDING_GENERATION_INVALID") from exc
|
|
if generation > 0:
|
|
with self.controller.store._connect() as connection:
|
|
earlier = connection.execute(
|
|
"SELECT 1 FROM jobs WHERE branch=? AND state=? "
|
|
"AND CAST(json_extract(request_json, '$.binding.generation') AS INTEGER) < ? LIMIT 1",
|
|
(row["branch"], JobState.RECONCILIATION_REQUIRED.value, generation),
|
|
).fetchone()
|
|
if earlier is not None:
|
|
raise ContinuationExecutionError("EARLIER_RECONCILIATION_REQUIRED")
|
|
self.credential.project()
|
|
try:
|
|
git_oid, api_oid, visible = self._observe_remote(row["branch"], new_oid)
|
|
if authority_oid == git_oid == api_oid == new_oid and visible:
|
|
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST")
|
|
return self._finish_remote_verification(job_id, new_oid, new_oid)
|
|
expected_oid = request["expected_branch_oid"]
|
|
if (
|
|
self.authority.is_ancestor(new_oid, authority_oid)
|
|
and git_oid is not None
|
|
and git_oid == api_oid
|
|
and self.authority.is_ancestor(git_oid, new_oid)
|
|
):
|
|
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_CHAIN")
|
|
self.remote.push_exact(row["branch"], git_oid, new_oid)
|
|
return self._finish_remote_verification(job_id, new_oid, new_oid)
|
|
if authority_oid == new_oid and git_oid == api_oid == expected_oid:
|
|
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_RETRY")
|
|
self.remote.push_exact(row["branch"], expected_oid, new_oid)
|
|
return self._finish_remote_verification(job_id, new_oid, new_oid)
|
|
return self.controller.store._receipt(row, duplicate=True).as_dict()
|
|
finally:
|
|
self.credential.revoke()
|