aios/aios
1
0
Fork 0
Code Issues Pull Requests 5 Actions Packages Projects Releases Wiki Activity

fix(gitops): bind post-advance jobs to exact head

Browse Source
This commit is contained in:
chatgpt-final-release-writer
2026-08-08 15:26:35 +08:00
parent a91bfcaecc
commit 1f787acede
3 changed files with 35 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
active-runtime/app/aios_gitea_machine_transport.py
+6 -10
View File
@@ -116,7 +116,7 @@ class GiteaMachineTransport:
"CREDENTIAL_FILE": str(credential),
}
def _push_new_branch(self, branch: str, new_oid: str) -> None:
def _push_branch_fast_forward(self, branch: str, new_oid: str) -> None:
askpass, additions = self._askpass()
environment = os.environ.copy()
environment.update(additions)
@@ -141,21 +141,17 @@ class GiteaMachineTransport:
if current == new_oid:
return new_oid
if current is None:
self._push_new_branch(branch, new_oid)
self._push_branch_fast_forward(branch, new_oid)
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CREATE_READBACK_MISMATCH")
return readback
if current != expected_oid:
raise GiteaTransportError("GITEA_BRANCH_EXPECTATION_MISMATCH")
owner, name = self.repository_id.split("/", 1)
status, value = self._request(
"PUT",
f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}",
{"new_commit_id": new_oid, "old_commit_id": expected_oid, "force": False},
)
if status not in (200, 201, 204):
raise GiteaTransportError("GITEA_BRANCH_CAS_FAILED")
# The preflight read binds the expected old tip. The Git receive-pack
# then enforces the actual fast-forward CAS while transferring the new
# objects; no force/lease override or arbitrary ref is permitted.
self._push_branch_fast_forward(branch, new_oid)
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CAS_READBACK_MISMATCH")
active-runtime/app/aios_gitops_job_executor.py
+27 -12
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import argparse
import base64
from dataclasses import dataclass
from dataclasses import dataclass, replace
import json
from pathlib import Path
from typing import Any, Mapping
@@ -44,12 +44,13 @@ class DurableJobExecutor:
return row, dict(request.get("binding") or {})
def _validate_campaign(self, request: Mapping[str, Any]) -> str:
operation = str(request.get("operation", ""))
expected_branch_oid = request.get("expected_branch_oid")
required = {
"task_id": TASK_ID,
"campaign_id": CAMPAIGN_ID,
"repository": REPOSITORY,
"branch": BRANCH,
"expected_branch_oid": EXPECTED_BRANCH_OID,
"authorization_recovery_digest": AUTHORIZATION_DIGEST,
}
for key, expected in required.items():
@@ -58,6 +59,12 @@ class DurableJobExecutor:
new_oid = request.get("new_branch_oid")
if not isinstance(new_oid, str) or len(new_oid) != 40 or any(ch not in "0123456789abcdef" for ch in new_oid):
raise JobExecutionError("NEW_BRANCH_OID_INVALID")
if not isinstance(expected_branch_oid, str) or len(expected_branch_oid) != 40 or any(ch not in "0123456789abcdef" for ch in expected_branch_oid):
raise JobExecutionError("EXPECTED_BRANCH_OID_INVALID")
if operation == "ADVANCE_EXISTING_CAMPAIGN_BRANCH" and expected_branch_oid == new_oid:
raise JobExecutionError("EXPECTED_BRANCH_OID_INVALID")
if operation != "ADVANCE_EXISTING_CAMPAIGN_BRANCH" and expected_branch_oid != new_oid:
raise JobExecutionError("POST_ADVANCE_HEAD_BINDING_INVALID")
try:
recovery = load_recovered_authorization(self.recovery_path)
except RecoveryEvidenceError as exc:
@@ -76,10 +83,12 @@ class DurableJobExecutor:
if recovery.get("expected_branch_oid") != EXPECTED_BRANCH_OID:
raise JobExecutionError("RECOVERY_BRANCH_OID_MISMATCH")
authority_head = self.authority.read_ref(f"refs/heads/{BRANCH}")
if authority_head not in {EXPECTED_BRANCH_OID, new_oid}:
if authority_head not in {expected_branch_oid, new_oid}:
raise JobExecutionError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH")
if not self.authority.ancestry_check(EXPECTED_BRANCH_OID, new_oid):
if not self.authority.ancestry_check(expected_branch_oid, new_oid):
raise JobExecutionError("NEW_OID_NOT_DESCENDANT")
if not self.authority.ancestry_check(EXPECTED_BRANCH_OID, new_oid):
raise JobExecutionError("NEW_OID_OUTSIDE_AUTHORIZED_LINEAGE")
identity = self.broker.verify_identity()
if identity.get("repository") != REPOSITORY or identity.get("login") == "aiman":
raise JobExecutionError("MACHINE_IDENTITY_SCOPE_INVALID")
@@ -93,7 +102,7 @@ class DurableJobExecutor:
action = "CREATE_REQUIRED"
elif remote == new_oid:
action = "ALREADY_IN_SYNC"
elif remote == EXPECTED_BRANCH_OID:
elif remote == request["expected_branch_oid"]:
action = "FAST_FORWARD_REQUIRED"
else:
action = "MANUAL_OWNER_REVIEW_REQUIRED"
@@ -103,7 +112,7 @@ class DurableJobExecutor:
"decision": "SAFE_TO_ADVANCE",
"authority_action": "CAS_REQUIRED",
"gitea_action": action,
"previous_branch_oid": EXPECTED_BRANCH_OID,
"previous_branch_oid": request["expected_branch_oid"],
"final_branch_oid": new_oid,
"force_push": False,
}
@@ -142,10 +151,13 @@ class DurableJobExecutor:
final = self.broker.push_branch(new_oid)
return {"operation": operation, "remote_verified": final == new_oid, "final_network_oid": final}
if operation == "CREATE_PR":
self._validate_campaign(request)
return dict(self.broker.create_pr("AIOS autonomous release control plane", "Durable campaign continuation; exact head and recovery evidence are bound in the Job."))
if operation == "PUBLISH_CHECK":
self._validate_campaign(request)
return dict(self.broker.status("success", "Durable campaign control-plane checks passed"))
if operation == "PROTECTED_MERGE":
self._validate_campaign(request)
number = int(request.get("pr_number", "0"))
return dict(self.broker.merge(number))
if operation == "RECONCILE_AUTHORITY_MAIN":
@@ -160,35 +172,38 @@ class DurableJobExecutor:
self._recover_requested(job_id)
row, request = self._request(job_id)
new_oid = self._validate_campaign(request)
expected_oid = request["expected_branch_oid"]
state = JobState(row["state"])
if state == JobState.REQUESTED:
receipt = self.controller.advance(job_id, JobState.AUTHORIZED, side_effect_state="AUTHORIZED", receipt={"authorization_digest": AUTHORIZATION_DIGEST, "task_id": TASK_ID})
row = self.controller.store.read(job_id)
if JobState(row["state"]) == JobState.AUTHORIZED:
self.controller.lease_job(job_id, lease_id=f"lease-{job_id}", lineage_anchor_oid=LINEAGE_ANCHOR, current_head_oid=EXPECTED_BRANCH_OID, generation=1, lease_expires_at="2099-01-01T00:00:00Z", expected_revision=int(row["revision"]))
self.controller.lease_job(job_id, lease_id=f"lease-{job_id}", lineage_anchor_oid=LINEAGE_ANCHOR, current_head_oid=expected_oid, generation=1, lease_expires_at="2099-01-01T00:00:00Z", expected_revision=int(row["revision"]))
row = self.controller.store.read(job_id)
if JobState(row["state"]) == JobState.LEASED:
self.controller.advance(job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY", receipt={"branch": BRANCH, "expected_oid": EXPECTED_BRANCH_OID})
self.controller.advance(job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY", receipt={"branch": BRANCH, "expected_oid": expected_oid})
row = self.controller.store.read(job_id)
if JobState(row["state"]) == JobState.WORKTREE_READY:
self.controller.advance(job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
row = self.controller.store.read(job_id)
authority_before = self.authority.read_ref(f"refs/heads/{BRANCH}")
if authority_before == EXPECTED_BRANCH_OID:
authority_oid = self.authority.advance_task_branch_CAS(f"refs/heads/{BRANCH}", EXPECTED_BRANCH_OID, new_oid)
if authority_before == expected_oid:
authority_oid = self.authority.advance_task_branch_CAS(f"refs/heads/{BRANCH}", expected_oid, new_oid)
elif authority_before == new_oid:
authority_oid = new_oid
else:
raise JobExecutionError("AUTHORITY_BRANCH_CONFLICT")
row = self.controller.store.read(job_id)
if JobState(row["state"]) == JobState.EXECUTING:
self.controller.advance(job_id, JobState.LOCAL_VERIFIED, side_effect_state="LOCAL_VERIFIED", receipt={"authority": {"branch": BRANCH, "previous_oid": EXPECTED_BRANCH_OID, "new_oid": new_oid, "CAS_success": True, "readback_oid": authority_oid}})
self.controller.advance(job_id, JobState.LOCAL_VERIFIED, side_effect_state="LOCAL_VERIFIED", receipt={"authority": {"branch": BRANCH, "previous_oid": expected_oid, "new_oid": new_oid, "CAS_success": True, "readback_oid": authority_oid}})
row = self.controller.store.read(job_id)
remote_before = self.broker.readback()
if remote_before != new_oid:
if remote_before not in {None, EXPECTED_BRANCH_OID}:
if remote_before not in {None, expected_oid}:
raise JobExecutionError("GITEA_BRANCH_CONFLICT")
if JobState(row["state"]) == JobState.LOCAL_VERIFIED:
if hasattr(self.broker, "receipt"):
self.broker.receipt = replace(self.broker.receipt, expected_oid=expected_oid)
self.controller.advance(job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS", receipt={"gitea": {"previous_oid": remote_before, "desired_oid": new_oid, "force_push": False}})
final_network = self.broker.push_branch(new_oid)
else:
tests/test_live_executor_contract.py
+2 -3
View File
@@ -159,7 +159,6 @@ def test_machine_transport_branch_cas_is_not_force_push(monkeypatch):
def read_branch(_self, _branch):
return NEW if calls else OLD
monkeypatch.setattr(GiteaMachineTransport, "read_branch", read_branch)
monkeypatch.setattr(GiteaMachineTransport, "_request", lambda _self, method, path, body=None: calls.append((method, path, body)) or (200, {}))
monkeypatch.setattr(GiteaMachineTransport, "_push_branch_fast_forward", lambda _self, branch, new: calls.append((branch, new)))
assert transport.compare_and_swap_branch(executor_module.BRANCH, OLD, NEW) == NEW
assert calls[0][0] == "PUT"
assert calls[0][2]["force"] is False
assert calls == [(executor_module.BRANCH, NEW)]