Compare commits

...
10 changed files with 2538 additions and 47 deletions
@@ -13,6 +13,8 @@ BRIDGE_OPERATIONS = (
"getGovernedWriteCapabilities",
"createGovernedTask",
"acquireWriterLease",
"bootstrapWriterLease",
"renewWriterLease",
"commitFileGoverned",
"commitFilesGoverned",
"getGitOpsJob",
@@ -31,6 +33,8 @@ BRIDGE_OPERATIONS = (
OPERATION_REQUIRED_FIELDS = {
"createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
"acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"bootstrapWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"renewWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
"commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
"getGitOpsJob": ("job_id",),
@@ -46,6 +50,49 @@ OPERATION_REQUIRED_FIELDS = {
"getControlPlaneSnapshot": (),
}
BRIDGE_OPERATION_CONTRACTS = {
"acquireWriterLease": {
"method": "POST",
"path": "/v2/writer-lease",
"required_fields": OPERATION_REQUIRED_FIELDS["acquireWriterLease"],
"response_fields": (
"task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
"create_target_branch", "worktree_mode", "cas_ref_update_required",
),
},
"bootstrapWriterLease": {
"method": "POST",
"path": "/v2/writer-lease/bootstrap",
"required_fields": OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"],
"response_fields": (
"task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
"create_target_branch", "worktree_mode", "cas_ref_update_required",
"recovered", "bootstrapped", "runtime_rehydrated",
"durable_branch_lease_present", "idempotent",
),
},
"renewWriterLease": {
"method": "POST",
"path": "/v2/writer-lease/renew",
"required_fields": OPERATION_REQUIRED_FIELDS["renewWriterLease"],
"response_fields": (
"task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
"recovered", "bootstrapped", "runtime_rehydrated",
"durable_branch_lease_present", "idempotent",
),
},
"commitFileGoverned": {
"method": "POST", "path": "/v2/commit-file",
"required_fields": OPERATION_REQUIRED_FIELDS["commitFileGoverned"],
"response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
},
"commitFilesGoverned": {
"method": "POST", "path": "/v2/commit-files",
"required_fields": OPERATION_REQUIRED_FIELDS["commitFilesGoverned"],
"response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
},
}
class RuntimeDriftError(RuntimeError):
pass
@@ -122,6 +169,14 @@ class BridgeV2Adapter:
"legacy_writes_enabled": False,
"governed_writes_enabled": True,
"operations": list(BRIDGE_OPERATIONS),
"operation_contracts": {
operation: {
**contract,
"required_fields": list(contract["required_fields"]),
"response_fields": list(contract["response_fields"]),
}
for operation, contract in BRIDGE_OPERATION_CONTRACTS.items()
},
"force_push": False,
"main_write": False,
"tag_write": False,
@@ -151,6 +206,8 @@ class BridgeV2Adapter:
def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
def bootstrapWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("bootstrapWriterLease", request)
def renewWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("renewWriterLease", request)
def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)
File diff suppressed because it is too large Load Diff
@@ -329,6 +329,7 @@ class ContinuationPermissions:
@dataclass(frozen=True, slots=True)
class ContinuationRuntimeBinding:
request: WriterLeaseRequestV2
job_id: str
workspace_path: Path
lease_not_before: datetime
lease_expires_at: datetime
@@ -389,13 +390,20 @@ class ContinuationExecutor:
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) -> int:
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:
row = connection.execute(
"SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
(branch,),
).fetchone()
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)
@@ -404,6 +412,7 @@ class ContinuationExecutor:
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,
@@ -413,7 +422,7 @@ class ContinuationExecutor:
"lease_seconds": request.lease_seconds,
}
def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
request = binding.request
normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
if normalized != {**self._request_mapping(request), "create_target_branch": False, "worktree_mode": "detached_at_expected_head", "cas_ref_update_required": True}:
@@ -426,7 +435,7 @@ class ContinuationExecutor:
raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
if now >= binding.lease_expires_at:
raise ContinuationExecutionError("LEASE_EXPIRED")
if request.generation <= self._max_generation(request.target_branch):
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")
@@ -452,39 +461,60 @@ class ContinuationExecutor:
"generation": request.generation,
}
def execute(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit) -> dict[str, object]:
payload = commit.payload()
submitted = self.controller.submit(self._job_request(binding, commit, payload), payload=payload)
current = self.controller.store.read(submitted.job_id)
if JobState(current["state"]) == JobState.COMPLETED:
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:
return self.controller.store._receipt(current, duplicate=True).as_dict()
self._validate_preflight(binding)
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)
return self.controller.advance(
binding.job_id,
JobState.COMPLETED,
side_effect_state="REMOTE_VERIFIED",
receipt=receipt,
).as_dict()
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(submitted.job_id)
row = self.controller.store.read(binding.job_id)
if JobState(row["state"]) == JobState.REQUESTED:
self.controller.advance(
submitted.job_id,
binding.job_id,
JobState.AUTHORIZED,
side_effect_state="AUTHORIZED",
receipt={"task_id": request.task_id, "original_grant_consumed": True},
)
row = self.controller.store.read(submitted.job_id)
row = self.controller.store.read(binding.job_id)
if JobState(row["state"]) == JobState.AUTHORIZED:
self.controller.lease_job(
submitted.job_id,
lease_id=f"lease-{submitted.job_id}",
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"]),
)
self.controller.advance(submitted.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
self.controller.advance(submitted.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
self.controller.advance(binding.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
self.controller.advance(binding.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
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")
@@ -492,17 +522,22 @@ class ContinuationExecutor:
if authority_oid != new_oid:
raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
self.controller.advance(
submitted.job_id,
binding.job_id,
JobState.LOCAL_VERIFIED,
side_effect_state="LOCAL_VERIFIED",
receipt={"authority_oid": authority_oid, "force_push": False},
)
try:
self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
except Exception:
return self._remote_pending(submitted.job_id, authority_oid, new_oid)
return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
return self._remote_pending(binding.job_id, authority_oid, new_oid)
return self._finish_remote_verification(
binding.job_id,
authority_oid,
new_oid,
before_complete=before_complete,
)
finally:
if projected:
self.credential.revoke()
@@ -525,7 +560,14 @@ class ContinuationExecutor:
)
return result.as_dict()
def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
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"]
@@ -542,8 +584,20 @@ class ContinuationExecutor:
"reconciliation_decision": decision.action.value,
"force_push": False,
}
self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt=receipt)
return self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt=receipt).as_dict()
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)
@@ -49,6 +49,7 @@ class WriterLeaseRequestV2:
target_branch: str
generation: int
lease_seconds: int = 900
job_id: str = ""
schema = "aios.task-writer-lease.v2"
accepted_schemas = frozenset({"aios.task-writer-lease.v2", "aios.task-writer-lease-request.v2"})
@@ -56,6 +57,8 @@ class WriterLeaseRequestV2:
def validate(self, probe: BranchProbe) -> None:
if not self.task_id or self.repository != "aios/aios":
raise ContinuationError("request_identity_invalid")
if not self.job_id or not self.job_id.startswith("job-"):
raise ContinuationError("job_id_invalid")
if self.branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
raise ContinuationError("branch_mode_invalid")
_oid(self.base_oid, "base_oid")
@@ -27,6 +27,19 @@ class ServiceUnavailable(JobError):
pass
@dataclass(frozen=True, slots=True)
class LeaseRenewalResult:
"""Durable outcome of one writer-lease recovery attempt."""
renewed: bool
idempotent: bool
durable: bool
lease_expires_at: str | None
def __bool__(self) -> bool:
return self.renewed or self.idempotent
class BindingConflict(JobError):
pass
@@ -66,6 +79,14 @@ def utc_now() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def parse_utc(value: str) -> datetime:
normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
raise ValueError("LEASE_EXPIRY_INVALID")
return parsed.astimezone(timezone.utc)
def canonical_json(value: Any) -> bytes:
return (
json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n"
@@ -404,13 +425,16 @@ class GitOpsJobController:
self.store._mirror(result)
return self.store._receipt(result, duplicate=True)
def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE") -> None:
def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE", lease_expires_at: str | None = None) -> None:
if len(current_head_oid) != 40 or generation < 1:
raise JobError("LEASE_BINDING_INVALID")
try:
with self.store._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
if lease_expires_at is None:
updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
else:
updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,lease_expires_at=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, lease_expires_at, state, lease_id, expected_revision))
if updated.rowcount != 1:
raise JobError("LEASE_REVISION_CAS_CONFLICT")
connection.commit()
@@ -419,6 +443,94 @@ class GitOpsJobController:
except sqlite3.Error as exc:
raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
def renew_lease(
self,
*,
branch: str,
lineage_anchor_oid: str,
current_head_oid: str,
generation: int,
lease_expires_at: str,
) -> LeaseRenewalResult:
if (
not branch
or len(lineage_anchor_oid) != 40
or len(current_head_oid) != 40
or generation < 1
or not lease_expires_at.strip()
):
raise JobError("LEASE_BINDING_INVALID")
try:
with self.store._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
rows = connection.execute(
"SELECT lease_id, revision, generation, state, lease_expires_at "
"FROM branch_leases "
"WHERE branch=? AND lineage_anchor_oid=? AND current_head_oid=? "
"AND generation=? AND state='ACTIVE' ORDER BY revision DESC",
(branch, lineage_anchor_oid, current_head_oid, generation),
).fetchall()
if not rows:
connection.commit()
return LeaseRenewalResult(False, False, False, None)
if len(rows) != 1:
raise JobError("LEASE_BINDING_CONFLICT")
row = rows[0]
try:
current_expires_at = parse_utc(str(row["lease_expires_at"]))
requested_expires_at = parse_utc(lease_expires_at)
except (TypeError, ValueError) as exc:
raise JobError("LEASE_EXPIRY_INVALID") from exc
now = datetime.now(timezone.utc)
if requested_expires_at <= now:
raise JobError("LEASE_EXPIRY_INVALID")
if current_expires_at > now:
connection.commit()
return LeaseRenewalResult(
False,
True,
True,
str(row["lease_expires_at"]),
)
updated = connection.execute(
"UPDATE branch_leases SET lease_expires_at=?,revision=revision+1 "
"WHERE lease_id=? AND revision=? AND generation=? "
"AND state='ACTIVE' AND lease_expires_at=?",
(
lease_expires_at,
row["lease_id"],
int(row["revision"]),
generation,
row["lease_expires_at"],
),
)
if updated.rowcount != 1:
current = connection.execute(
"SELECT lease_expires_at, generation, state FROM branch_leases "
"WHERE lease_id=?",
(row["lease_id"],),
).fetchone()
if (
current is not None
and int(current["generation"]) == generation
and current["state"] == "ACTIVE"
and parse_utc(str(current["lease_expires_at"])) > now
):
connection.commit()
return LeaseRenewalResult(
False,
True,
True,
str(current["lease_expires_at"]),
)
raise JobError("LEASE_REVISION_CAS_CONFLICT")
connection.commit()
return LeaseRenewalResult(True, False, True, lease_expires_at)
except JobError:
raise
except sqlite3.Error as exc:
raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
def advance(self, job_id: str, state: JobState, *, side_effect_state: str, receipt: Mapping[str, Any] | None = None, expected_revision: int | None = None) -> JobReceipt:
return self._advance(
job_id,
@@ -964,6 +964,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
required = {
"schema",
"task_id",
"job_id",
"repository",
"branch_mode",
"base_oid",
@@ -976,6 +977,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
raise ProvisioningError("LEASE_REQUEST_V2_FIELDS_REJECTED", 409)
request = WriterLeaseRequestV2(
task_id=str(raw["task_id"]),
job_id=str(raw["job_id"]),
repository=str(raw["repository"]),
branch_mode=str(raw["branch_mode"]),
base_oid=str(raw["base_oid"]),
@@ -988,6 +990,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
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,
+223
View File
@@ -240,6 +240,114 @@ components:
additionalProperties: false
properties:
candidate_id: {type: string}
BridgeWriterLeaseRecoveryRequest:
type: object
additionalProperties: false
required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds]
properties:
schema:
type: string
enum: [aios.task-writer-lease.v2, aios.task-writer-lease-request.v2]
task_id: {type: string, minLength: 1, maxLength: 256}
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
repository: {type: string, const: aios/aios}
branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
lease_seconds: {type: integer, minimum: 1, maximum: 86400}
BridgeWriterLeaseResponse:
type: object
additionalProperties: false
required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, create_target_branch, worktree_mode, cas_ref_update_required]
properties:
schema: {type: string, enum: [aios.task-writer-lease.v2]}
task_id: {type: string}
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
repository: {type: string, const: aios/aios}
branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
lease_seconds: {type: integer, minimum: 1, maximum: 86400}
create_target_branch: {type: boolean}
worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
cas_ref_update_required: {type: boolean, const: true}
lease_expires_at: {type: string, format: date-time}
BridgeWriterLeaseRecoveryResponse:
type: object
additionalProperties: false
required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, lease_expires_at, create_target_branch, worktree_mode, cas_ref_update_required, recovered, bootstrapped, idempotent, runtime_rehydrated, durable_branch_lease_present]
properties:
schema: {type: string, enum: [aios.task-writer-lease.v2]}
task_id: {type: string}
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
repository: {type: string, const: aios/aios}
branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
lease_seconds: {type: integer, minimum: 1, maximum: 86400}
lease_expires_at: {type: string, format: date-time}
create_target_branch: {type: boolean, const: false}
worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
cas_ref_update_required: {type: boolean, const: true}
recovered: {type: boolean}
bootstrapped: {type: boolean}
idempotent: {type: boolean}
runtime_rehydrated: {type: boolean}
durable_branch_lease_present: {type: boolean}
BridgeCommitFileRequest:
type: object
additionalProperties: false
required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, path, content_base64, message]
properties:
task_id: {type: string, minLength: 1}
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
repository: {type: string, const: aios/aios}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
idempotency_key: {type: string, minLength: 1, maxLength: 256}
path: {type: string, minLength: 1, maxLength: 512}
content_base64: {type: string, format: byte}
message: {type: string, minLength: 1, maxLength: 200}
BridgeCommitFilesRequest:
type: object
additionalProperties: false
required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message]
properties:
task_id: {type: string, minLength: 1}
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
repository: {type: string, const: aios/aios}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
idempotency_key: {type: string, minLength: 1, maxLength: 256}
files: {type: object, additionalProperties: {type: string, format: byte}, minProperties: 1}
message: {type: string, minLength: 1, maxLength: 200}
BridgeCommitResponse:
type: object
additionalProperties: true
required: [job_id, commit_oid, commit_visible, branch_mode, target_branch, generation]
properties:
job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
commit_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
commit_visible: {type: boolean}
branch_mode: {type: string, enum: [CONTINUE_EXISTING]}
target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
generation: {type: integer, minimum: 1}
BridgeGovernedCapabilities:
type: object
required: [bridge_schema_version, governed_writes_enabled, operations, operation_contracts]
properties:
bridge_schema_version: {type: string, const: aios.chatgpt-gitops-bridge.v2}
governed_writes_enabled: {type: boolean, const: true}
operations: {type: array, items: {type: string}}
operation_contracts: {type: object, additionalProperties: true}
paths:
/createReleaseIntent:
post:
@@ -567,3 +675,118 @@ paths:
'200': {description: Safe GitOps plan}
'401': {description: Unauthorized}
x-openai-isConsequential: true
/v2/capabilities:
get:
operationId: getGovernedWriteCapabilities
summary: Read the governed Bridge v2 capability contract
responses:
'200':
description: Governed operations and their request/response contracts
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeGovernedCapabilities'}
/v2/writer-lease:
post:
operationId: acquireWriterLease
summary: Acquire the exact governed writer lease for a task generation
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
responses:
'200':
description: Exact detached worktree lease binding
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseResponse'}
'400': {description: Invalid lease binding}
'409': {description: Branch, base, or generation conflict}
x-openai-isConsequential: true
/v2/writer-lease/bootstrap:
post:
operationId: bootstrapWriterLease
summary: Bootstrap a governed writer runtime without creating a durable lease
description: >
Staging-only runtime recovery. It validates the exact task, repository,
base OID, expected branch OID, target branch, and generation. It must
not create a branch, insert a branch lease, or consume a generation.
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
responses:
'200':
description: Runtime binding bootstrapped or returned idempotently
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
'400': {description: Invalid lease binding}
'409': {description: A matching durable lease already exists or binding conflicts}
'503': {description: Runtime or durable state unavailable}
x-openai-isConsequential: true
/v2/writer-lease/renew:
post:
operationId: renewWriterLease
summary: Recover an expired governed writer lease without changing its binding
description: >
Renews only the current task generation and exact repository, base OID,
expected branch OID, and target branch. Durable renewal is revision-CAS
protected; runtime rehydration is reported separately.
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
responses:
'200':
description: Lease recovered or idempotently observed as already active
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
'400': {description: Invalid schema, generation, or lease duration}
'409': {description: Task, base, repository, branch, generation, or durable lease conflict}
'503': {description: Durable job store or runtime unavailable}
x-openai-isConsequential: true
/v2/commit-file:
post:
operationId: commitFileGoverned
summary: Commit one file through the governed writer lease
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeCommitFileRequest'}
responses:
'200':
description: Authority and remote visibility receipt
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
'409': {description: Lease, expected-head, idempotency, or repository conflict}
'503': {description: Reconciliation required}
x-openai-isConsequential: true
/v2/commit-files:
post:
operationId: commitFilesGoverned
summary: Atomically commit a bounded set of files through the governed writer lease
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeCommitFilesRequest'}
responses:
'200':
description: Authority and remote visibility receipt
content:
application/json:
schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
'409': {description: Lease, expected-head, idempotency, or repository conflict}
'503': {description: Reconciliation required}
x-openai-isConsequential: true
+122 -1
View File
@@ -2,6 +2,7 @@ import base64
from pathlib import Path
from aios_authority_git_executor import AuthorityGitExecutor
from aios_chatgpt_bridge_v2 import BridgeV2Adapter, OPERATION_REQUIRED_FIELDS
from aios_chatgpt_bridge_v2_server import BridgeHandler, RuntimeConfig
from aios_continuation_executor import GiteaContinuationRemote, ServerSideCredentialProjection
from aios_continuation_lease import WriterLeaseRequestV2
@@ -26,6 +27,7 @@ class Probe:
def test_create_new_lease_is_exact_base_and_absent():
request = WriterLeaseRequestV2(
job_id="job-bridge-runtime-lineage-test",
task_id="TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
repository="aios/aios",
branch_mode="CREATE_NEW",
@@ -39,6 +41,7 @@ def test_create_new_lease_is_exact_base_and_absent():
def test_create_new_lease_rejects_existing_target():
request = WriterLeaseRequestV2(
job_id="job-bridge-runtime-lineage-test",
task_id="task",
repository="aios/aios",
branch_mode="CREATE_NEW",
@@ -75,7 +78,7 @@ def test_slash_branch_uses_paginated_collection():
def test_http_route_contract_is_complete():
expected = {
"/v2/runtime", "/v2/capabilities", "/v2/governed-task",
"/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
"/v2/writer-lease", "/v2/writer-lease/bootstrap", "/v2/writer-lease/renew", "/v2/commit-file", "/v2/commit-files",
"/v2/job", "/v2/job/resume", "/v2/job/reconcile",
"/v2/tag/approval-request",
"/v2/tag/publish",
@@ -88,3 +91,121 @@ def test_http_route_contract_is_complete():
from aios_chatgpt_bridge_v2_server import ROUTES
assert set(ROUTES) == expected
def test_renew_writer_lease_is_exposed_in_action_and_capability_contracts():
adapter = BridgeV2Adapter(runtime_status={})
capabilities = adapter.getGovernedWriteCapabilities()
contract = capabilities["operation_contracts"]["renewWriterLease"]
assert contract["method"] == "POST"
assert contract["path"] == "/v2/writer-lease/renew"
assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["renewWriterLease"]
assert "runtime_rehydrated" in contract["response_fields"]
action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
assert "/v2/writer-lease/renew:" in action_schema
assert "operationId: renewWriterLease" in action_schema
assert "BridgeWriterLeaseRecoveryRequest" in action_schema
assert "BridgeWriterLeaseRecoveryResponse" in action_schema
assert {
capabilities["operation_contracts"][operation]["path"]
for operation in (
"acquireWriterLease",
"renewWriterLease",
"commitFileGoverned",
"commitFilesGoverned",
)
} == {
"/v2/writer-lease",
"/v2/writer-lease/renew",
"/v2/commit-file",
"/v2/commit-files",
}
def test_bootstrap_writer_lease_is_exposed_in_action_and_capability_contracts():
adapter = BridgeV2Adapter(runtime_status={})
capabilities = adapter.getGovernedWriteCapabilities()
contract = capabilities["operation_contracts"]["bootstrapWriterLease"]
assert contract["method"] == "POST"
assert contract["path"] == "/v2/writer-lease/bootstrap"
assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"]
assert "bootstrapped" in contract["response_fields"]
assert "durable_branch_lease_present" in contract["response_fields"]
action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
assert "/v2/writer-lease/bootstrap:" in action_schema
assert "operationId: bootstrapWriterLease" in action_schema
assert "bootstrapped: {type: boolean}" in action_schema
assert "durable_branch_lease_present: {type: boolean}" in action_schema
def test_http_allows_explicit_job_id_on_lease_and_commit_requests():
handler = BridgeHandler.__new__(BridgeHandler)
lease_body = {
"schema": WriterLeaseRequestV2.schema,
"task_id": "task",
"job_id": "job-explicit-runtime-lineage",
"repository": "aios/aios",
"branch_mode": "CONTINUE_EXISTING",
"base_oid": "a" * 40,
"expected_branch_oid": "a" * 40,
"target_branch": "agent/explicit-runtime-lineage",
"generation": 2,
"lease_seconds": 120,
}
handler._allowed("acquireWriterLease", lease_body)
handler._allowed(
"commitFilesGoverned",
{
"task_id": "task",
"job_id": "job-explicit-runtime-lineage",
"repository": "aios/aios",
"expected_branch_oid": "a" * 40,
"target_branch": "agent/explicit-runtime-lineage",
"generation": 2,
"idempotency_key": "commit-explicit-runtime-lineage",
"files": {"docs/x.txt": "eA=="},
"message": "test",
},
)
def test_http_allows_omitted_job_id_for_durable_server_resolution():
handler = BridgeHandler.__new__(BridgeHandler)
lease_body = {
"schema": WriterLeaseRequestV2.schema,
"task_id": "task",
"repository": "aios/aios",
"branch_mode": "CONTINUE_EXISTING",
"base_oid": "a" * 40,
"expected_branch_oid": "a" * 40,
"target_branch": "agent/implicit-runtime-lineage",
"generation": 2,
"lease_seconds": 120,
}
handler._allowed("acquireWriterLease", lease_body)
handler._allowed(
"commitFilesGoverned",
{
"task_id": "task",
"repository": "aios/aios",
"expected_branch_oid": "a" * 40,
"target_branch": "agent/implicit-runtime-lineage",
"generation": 2,
"idempotency_key": "commit-implicit-runtime-lineage",
"files": {"docs/x.txt": "eA=="},
"message": "test",
},
)
def test_capability_contracts_return_exact_job_id_lineage():
contracts = BridgeV2Adapter(runtime_status={}).getGovernedWriteCapabilities()["operation_contracts"]
for operation in ("acquireWriterLease", "bootstrapWriterLease", "renewWriterLease"):
assert "job_id" in contracts[operation]["response_fields"]
for operation in ("commitFileGoverned", "commitFilesGoverned"):
assert "job_id" in contracts[operation]["response_fields"]
@@ -0,0 +1,175 @@
from __future__ import annotations
from pathlib import Path
import re
import json
from types import SimpleNamespace
import pytest
import yaml
from aios_chatgpt_bridge_v2_server import GovernedRuntime
from aios_continuation_executor import ContinuationExecutor
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import JobState
TASK = "continuation-normalization-test"
JOB = "job-continuation-normalization-test"
REPO = "aios/aios"
BASE = "a" * 40
BRANCH = "agent/continuation-normalization-test"
class Authority:
def __init__(self, head: str | None) -> None:
self.head = head
def read_ref(self, ref: str) -> str | None:
assert ref == "refs/heads/" + BRANCH
return self.head
def lease(**overrides: object) -> WriterLeaseRequestV2:
values = {
"task_id": TASK,
"job_id": JOB,
"repository": REPO,
"branch_mode": "CREATE_NEW",
"base_oid": BASE,
"expected_branch_oid": BASE,
"target_branch": BRANCH,
"generation": 1,
"lease_seconds": 120,
}
values.update(overrides)
return WriterLeaseRequestV2(**values)
def runtime(*, head: str | None = BASE, **binding_overrides: object) -> GovernedRuntime:
value = GovernedRuntime.__new__(GovernedRuntime)
value.authority = Authority(head)
binding = {
"task_id": TASK,
"job_id": JOB,
"repository": REPO,
"base_oid": BASE,
"expected_branch_oid": BASE,
"target_branch": BRANCH,
"generation": 1,
"branch_mode": "CREATE_NEW",
}
binding.update(binding_overrides)
value.bindings = {TASK: binding}
return value
def test_same_lineage_create_new_normalizes_to_continue_existing() -> None:
normalized = runtime()._normalize_runtime_lease(lease())
assert normalized.branch_mode == "CONTINUE_EXISTING"
assert normalized.task_id == TASK
assert normalized.job_id == JOB
assert normalized.repository == REPO
assert normalized.target_branch == BRANCH
assert normalized.expected_branch_oid == BASE
assert normalized.generation == 1
def test_unprovisioned_create_new_stays_create_new() -> None:
assert runtime(head=None)._normalize_runtime_lease(lease()).branch_mode == "CREATE_NEW"
@pytest.mark.parametrize(
("field", "value"),
(
("repository", "other/repo"),
("base_oid", "b" * 40),
("expected_branch_oid", "b" * 40),
("target_branch", "agent/foreign"),
("generation", 2),
),
)
def test_binding_mismatch_fails_closed(field: str, value: object) -> None:
with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
runtime(**{field: value})._normalize_runtime_lease(lease())
def test_drifted_existing_head_fails_closed() -> None:
with pytest.raises(RuntimeError, match="EXPECTED_BRANCH_OID_CONFLICT"):
runtime(head="b" * 40)._normalize_runtime_lease(lease())
def test_openapi_contract_parses_and_requires_exact_commit_lineage() -> None:
document = yaml.safe_load(
Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
)
schema = document["components"]["schemas"]["BridgeCommitResponse"]
assert {
"job_id",
"commit_oid",
"commit_visible",
"branch_mode",
"target_branch",
"generation",
} <= set(schema["required"])
def test_changed_production_files_pass_secret_scan() -> None:
paths = [
Path("active-runtime/app/aios_chatgpt_bridge_v2_server.py"),
Path("active-runtime/app/aios_continuation_executor.py"),
Path("active-runtime/app/aios_continuation_lease.py"),
Path("active-runtime/app/authorized_task_writer_provisioning.py"),
Path("active-runtime/app/aios_chatgpt_bridge_v2.py"),
Path("docs/gitops/openapi-chatgpt-actions.yaml"),
]
expression = re.compile(
r"(?i)(BEGIN (RSA|OPENSSH|EC) PRIVATE KEY|"
r"authorization:\\s*bearer\\s+[A-Za-z0-9._-]{12,}|"
r"api[_-]?key\\s*[:=]\\s*['\\\"][^'\\\"]{12,})"
)
hits = []
for path in paths:
for number, line in enumerate(
path.read_text(encoding="utf-8", errors="replace").splitlines(), 1
):
if expression.search(line):
hits.append(f"{path}:{number}")
assert hits == []
def test_remote_verified_job_is_not_completed_when_head_cas_callback_fails() -> None:
class Store:
def read(self, _job_id):
return {
"branch": BRANCH,
"request_json": json.dumps(
{"binding": {"expected_branch_oid": BASE}}
),
}
class Controller:
def __init__(self):
self.store = Store()
self.states = []
def advance(self, _job_id, state, **_kwargs):
self.states.append(state)
return SimpleNamespace(as_dict=lambda: {"state": state.value})
executor = ContinuationExecutor.__new__(ContinuationExecutor)
executor.controller = Controller()
executor._observe_remote = lambda _branch, oid: (oid, oid, True)
def fail_cas(_oid):
raise RuntimeError("LEASE_CAS_CONFLICT")
with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
executor._finish_remote_verification(
JOB,
"b" * 40,
"b" * 40,
before_complete=fail_cas,
)
assert executor.controller.states == [JobState.REMOTE_VERIFIED]
+708
View File
@@ -0,0 +1,708 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
import hashlib
from dataclasses import asdict
from pathlib import Path
import sqlite3
import json
from types import SimpleNamespace
import threading
import pytest
from aios_chatgpt_bridge_v2_server import GovernedRuntime
from aios_continuation_executor import ContinuationPermissions, ContinuationRuntimeBinding
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
from aios_governed_task_bootstrap import GovernedTaskBootstrap
TASK_ID = "gov-writer-lease-recovery-test"
BRANCH = "agent/writer-lease-recovery-test"
BASE = "a" * 40
JOB_ID_PREFIX = "writer-lease-runtime-lineage"
class LocalAuthority:
def __init__(self, head: str | None = None) -> None:
self.head = head
def read_ref(self, _ref: str) -> str | None:
return self.head
def commit_exists(self, _oid: str) -> bool:
return True
def ancestry_check(self, _older: str, _newer: str) -> bool:
return True
def writer_request(task_id: str = TASK_ID, **overrides: object) -> dict[str, object]:
request: dict[str, object] = {
"schema": WriterLeaseRequestV2.schema,
"task_id": task_id,
"repository": "aios/aios",
"branch_mode": "CREATE_NEW",
"base_oid": BASE,
"expected_branch_oid": BASE,
"target_branch": BRANCH,
"generation": 1,
"lease_seconds": 120,
}
request.update(overrides)
return request
def make_runtime(tmp_path: Path, task_id: str = TASK_ID, *, authority_head: str | None = None) -> GovernedRuntime:
runtime = GovernedRuntime.__new__(GovernedRuntime)
runtime.config = SimpleNamespace(repository="aios/aios", state_root=tmp_path)
runtime.authority = LocalAuthority(authority_head)
runtime.bindings = {
task_id: {
"task_id": task_id,
"repository": "aios/aios",
"base_oid": BASE,
"expected_branch_oid": BASE,
"target_branch": BRANCH,
"branch_mode": "CREATE_NEW",
"generation": 1,
}
}
runtime.runtimes = {}
runtime.controllers = {}
runtime.bootstrap = GovernedTaskBootstrap(
runtime.authority,
tmp_path,
remote_name="test",
token_path=tmp_path / "token",
ssh_key_path=tmp_path / "key",
)
runtime.bindings = {}
runtime.create_task(
{
"task_id": task_id,
"repository": "aios/aios",
"base_oid": BASE,
"expected_branch_oid": BASE,
"target_branch": BRANCH,
"risk": "engineering",
"description": "governed runtime lineage test",
"idempotency_key": JOB_ID_PREFIX + "-" + task_id,
}
)
job_root = runtime._job_root(task_id)
runtime.controllers[task_id] = GitOpsJobController(
JobStore(job_root, payload_root=job_root / "payloads")
)
def build(
lease: WriterLeaseRequestV2,
*,
job_id: str,
lease_not_before: datetime | None = None,
lease_expires_at: datetime | None = None,
) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
controller = runtime.controllers.get(lease.task_id)
if controller is None:
root = runtime._job_root(lease.task_id)
controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
now = datetime.now(timezone.utc)
return (
ContinuationRuntimeBinding(
request=lease,
job_id=job_id,
workspace_path=tmp_path / "worktrees" / lease.task_id,
lease_not_before=lease_not_before or now - timedelta(seconds=1),
lease_expires_at=lease_expires_at or now + timedelta(seconds=lease.lease_seconds),
original_receipt_valid=True,
original_grant_consumed=True,
permissions=ContinuationPermissions(),
),
controller,
object(),
)
runtime._build_runtime = build
return runtime
def create_durable_lease(
runtime: GovernedRuntime,
*,
expires_at: str = "2020-01-01T00:00:00+00:00",
) -> GitOpsJobController:
root = runtime._job_root(TASK_ID)
controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
receipt = SimpleNamespace(job_id=runtime.bindings[TASK_ID]["job_id"])
controller.lease_job(
receipt.job_id,
lease_id="writer-lease-recovery",
lineage_anchor_oid=BASE,
current_head_oid=BASE,
generation=1,
lease_expires_at=expires_at,
expected_revision=0,
)
runtime.controllers[TASK_ID] = controller
return controller
def durable_row(runtime: GovernedRuntime) -> tuple[str, int]:
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
row = connection.execute(
"SELECT lease_expires_at, revision FROM branch_leases WHERE lease_id=?",
("writer-lease-recovery",),
).fetchone()
assert row is not None
return row
def test_first_bootstrap_refreshes_only_runtime_without_durable_lease(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
result = runtime.bootstrap_writer_lease(writer_request())
assert result["bootstrapped"] is True
assert result["runtime_rehydrated"] is False
assert result["durable_branch_lease_present"] is False
assert result["idempotent"] is False
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
def test_duplicate_bootstrap_is_idempotent_without_consuming_generation(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
first = runtime.bootstrap_writer_lease(writer_request())
second = runtime.bootstrap_writer_lease(writer_request())
assert first["bootstrapped"] is True
assert first["generation"] == 1
assert second["bootstrapped"] is False
assert second["idempotent"] is True
assert second["generation"] == 1
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
def test_commit_after_bootstrap_reuses_the_exact_runtime_binding(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
class RecordingWorkspace:
def __init__(self) -> None:
self.files: dict[str, bytes] = {}
def materialize_files(self, files: dict[str, bytes]) -> None:
self.files = dict(files)
class RecordingExecutor:
def __init__(self) -> None:
self.workspace = RecordingWorkspace()
self.binding = None
self.commit = None
def execute(self, binding: object, commit: object) -> dict[str, object]:
self.binding = binding
self.commit = commit
return {'task_id': binding.request.task_id, 'generation': binding.request.generation, 'files': sorted(self.workspace.files)}
executor = RecordingExecutor()
original_build = runtime._build_runtime
def build(lease: WriterLeaseRequestV2, *, job_id: str, lease_not_before: datetime | None=None, lease_expires_at: datetime | None=None) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
binding, controller, _ = original_build(lease, job_id=job_id, lease_not_before=lease_not_before, lease_expires_at=lease_expires_at)
return (binding, controller, executor)
runtime._build_runtime = build
runtime.bootstrap_writer_lease(writer_request())
lease_response = runtime.acquire_lease(writer_request())
binding, controller, executor = runtime.runtimes[TASK_ID, 1]
assert lease_response['job_id'] == binding.job_id
assert lease_response['durable_branch_lease_present'] is True
_original_execute = executor.execute
def _execute_with_governed_commit_contract(*args, **kwargs):
_before_complete = kwargs.pop('before_complete', None)
_legacy = dict(_original_execute(*args, **kwargs))
_result = {**_legacy, 'job_id': binding.job_id, 'state': 'COMPLETED', 'side_effect_state': 'REMOTE_VERIFIED', 'receipt': {'authority_oid': 'b' * 40, 'git_remote_oid': 'b' * 40, 'api_branch_oid': 'b' * 40, 'commit_visible': True, 'remote_verified': True, 'force_push': False}}
if _before_complete is not None:
_before_complete('b' * 40)
return _result
executor.execute = _execute_with_governed_commit_contract
_bound_job_id = binding.job_id
result = runtime.commit_files({'task_id': TASK_ID, 'repository': 'aios/aios', 'generation': 1, 'target_branch': BRANCH, 'expected_branch_oid': BASE, 'idempotency_key': 'commit-after-bootstrap-v1', 'path': 'docs/recovery.txt', 'content_base64': 'YQ==', 'message': 'test commit after bootstrap'})
assert result['task_id'] == TASK_ID
assert result['generation'] == 1
assert result['files'] == ['docs/recovery.txt']
assert executor.binding.request.task_id == TASK_ID
assert executor.binding.request.generation == 1
database = runtime._job_root(TASK_ID) / 'state.sqlite3'
with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
assert connection.execute('SELECT COUNT(*) FROM branch_leases').fetchone()[0] == 1
assert connection.execute('PRAGMA foreign_key_check(branch_leases)').fetchall() == []
assert result['commit_oid'] == 'b' * 40
assert result['commit_visible'] is True
assert result['branch_mode'] == 'CONTINUE_EXISTING'
assert result['target_branch'] == BRANCH
assert int(result['generation']) == 1
with controller.store._connect() as connection:
_persisted = connection.execute('SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
assert _persisted is not None
assert _persisted[0] == 'b' * 40
assert int(_persisted[1]) == 1
fresh_runtime = make_runtime(tmp_path, authority_head='b' * 40)
fresh_response = fresh_runtime.bootstrap_writer_lease(writer_request(branch_mode='CONTINUE_EXISTING', expected_branch_oid='b' * 40))
fresh_binding, _, _ = fresh_runtime.runtimes[TASK_ID, 1]
assert fresh_response['job_id'] == str(_bound_job_id)
assert fresh_response['runtime_rehydrated'] is True
assert fresh_binding.job_id == str(_bound_job_id)
assert fresh_binding.request.expected_branch_oid == 'b' * 40
with fresh_runtime.controllers[TASK_ID].store._connect() as connection:
_fresh_lease = connection.execute('SELECT job_id,current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
assert tuple(_fresh_lease) == (str(_bound_job_id), 'b' * 40, 1)
@pytest.mark.parametrize(
("field", "value", "error"),
(
("repository", "other/repository", "LEASE_BINDING_CONFLICT"),
("base_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
("expected_branch_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
("target_branch", "agent/wrong-target", "LEASE_BINDING_CONFLICT"),
("generation", 2, "GENERATION_BINDING_CONFLICT"),
),
)
def test_bootstrap_rejects_any_nonmatching_binding_field(
tmp_path: Path, field: str, value: object, error: str
) -> None:
runtime = make_runtime(tmp_path)
with pytest.raises(RuntimeError, match=error):
runtime.bootstrap_writer_lease(writer_request(**{field: value}))
def test_bootstrap_rejects_unknown_task_ownership(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
with pytest.raises(RuntimeError, match="TASK_OWNERSHIP_INVALID"):
runtime.bootstrap_writer_lease(writer_request(task_id="unknown-task"))
def test_renew_missing_durable_lease_is_not_bootstrap(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
with pytest.raises(RuntimeError, match="LEASE_NOT_FOUND"):
runtime.renew_writer_lease(writer_request())
def test_renew_is_generation_scoped_before_durable_lookup(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
create_durable_lease(runtime)
with pytest.raises(RuntimeError, match="GENERATION_BINDING_CONFLICT"):
runtime.renew_writer_lease(writer_request(generation=2))
assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 0)
def test_restart_rehydrates_only_from_matching_durable_lease(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
create_durable_lease(runtime)
runtime.runtimes.clear()
result = runtime.renew_writer_lease(writer_request())
expires_at, revision = durable_row(runtime)
assert result["recovered"] is True
assert result["runtime_rehydrated"] is True
assert result["durable_branch_lease_present"] is True
assert revision == 1
assert datetime.fromisoformat(expires_at) > datetime.now(timezone.utc)
def test_duplicate_renewal_is_idempotent_after_first_durable_recovery(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
create_durable_lease(runtime)
first = runtime.renew_writer_lease(writer_request())
second = runtime.renew_writer_lease(writer_request())
assert first["recovered"] is True
assert first["idempotent"] is False
assert second["recovered"] is False
assert second["idempotent"] is True
assert durable_row(runtime)[1] == 1
def test_update_lease_keeps_expiry_without_optional_argument_and_keeps_cas(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
controller = create_durable_lease(runtime)
controller.update_lease(
"writer-lease-recovery",
expected_revision=0,
current_head_oid=BASE,
generation=1,
)
assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 1)
with pytest.raises(JobError, match="LEASE_REVISION_CAS_CONFLICT"):
controller.update_lease(
"writer-lease-recovery",
expected_revision=0,
current_head_oid=BASE,
generation=1,
)
def test_two_runtime_connections_resolve_renewal_cas_race_idempotently(tmp_path: Path) -> None:
first = make_runtime(tmp_path)
first_controller = create_durable_lease(first)
second = make_runtime(tmp_path)
second_controller = GitOpsJobController(
JobStore(second._job_root(TASK_ID), payload_root=second._job_root(TASK_ID) / "payloads")
)
second.controllers[TASK_ID] = second_controller
gate = threading.Barrier(2)
def gate_update(controller: GitOpsJobController) -> None:
original = controller.update_lease
def update(*args: object, **kwargs: object) -> None:
gate.wait(timeout=5)
original(*args, **kwargs)
controller.update_lease = update
gate_update(first_controller)
gate_update(second_controller)
with ThreadPoolExecutor(max_workers=2) as pool:
results = list(pool.map(lambda runtime: runtime.renew_writer_lease(writer_request()), (first, second)))
assert sorted((item["recovered"], item["idempotent"]) for item in results) == [
(False, True),
(True, False),
]
assert durable_row(first)[1] == 1
def test_bootstrap_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
response = runtime.bootstrap_writer_lease(writer_request())
binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
assert response["job_id"] == expected_job_id
assert binding.job_id == expected_job_id
assert binding.request.job_id == expected_job_id
def test_restart_rehydrate_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
first = make_runtime(tmp_path)
expected_job_id = first.bindings[TASK_ID]["job_id"]
first.acquire_lease(writer_request())
runtime = make_runtime(tmp_path)
assert runtime is not first
assert runtime.runtimes == {}
response = runtime.bootstrap_writer_lease(writer_request())
binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
assert response["job_id"] == expected_job_id
assert binding.job_id == expected_job_id
assert binding.request.job_id == expected_job_id
def test_wrong_governed_job_id_is_denied_before_runtime_or_lease_side_effect(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
runtime.bootstrap_writer_lease(writer_request(job_id="job-wrong-lineage"))
assert runtime.runtimes == {}
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
def test_acquire_creates_exact_governed_writer_lease_with_fk_and_cas(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
response = runtime.acquire_lease(writer_request())
assert response["job_id"] == expected_job_id
assert response["durable_branch_lease_present"] is True
assert response["idempotent"] is False
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
foreign_keys = connection.execute(
"PRAGMA foreign_key_list(branch_leases)"
).fetchall()
row = connection.execute(
"SELECT job_id,branch,lineage_anchor_oid,current_head_oid,generation,state,revision "
"FROM branch_leases"
).fetchone()
job = connection.execute(
"SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
(expected_job_id,),
).fetchone()
assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
assert any(
fk[2] == "jobs" and fk[3] == "job_id" and fk[4] == "job_id"
for fk in foreign_keys
)
assert row == (expected_job_id, BRANCH, BASE, BASE, 1, "ACTIVE", 0)
assert job == ("LEASED", "LEASED", 2)
def test_duplicate_acquire_observes_same_durable_writer_lease(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
first = runtime.acquire_lease(writer_request())
second = runtime.acquire_lease(writer_request())
assert first["job_id"] == expected_job_id
assert second["job_id"] == expected_job_id
assert second["idempotent"] is True
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
leases = connection.execute(
"SELECT lease_id,job_id,revision FROM branch_leases"
).fetchall()
job = connection.execute(
"SELECT state,revision FROM jobs WHERE job_id=?",
(expected_job_id,),
).fetchone()
assert leases == [("lease-" + expected_job_id, expected_job_id, 0)]
assert job == ("LEASED", 2)
def test_wrong_job_acquire_is_denied_before_authorize_or_lease(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
runtime.acquire_lease(writer_request(job_id="job-wrong-lineage"))
database = runtime._job_root(TASK_ID) / "state.sqlite3"
with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
job = connection.execute(
"SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
(expected_job_id,),
).fetchone()
lease_count = connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0]
assert job == ("REQUESTED", "NONE", 0)
assert lease_count == 0
def test_omitted_request_job_id_resolves_the_exact_durable_parent(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
raw = dict(writer_request())
response = runtime.acquire_lease(raw)
binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
assert response["job_id"] == expected_job_id
assert binding.job_id == expected_job_id
assert binding.request.job_id == expected_job_id
def test_durable_job_expected_head_mismatch_is_denied(tmp_path: Path) -> None:
runtime = make_runtime(tmp_path)
expected_job_id = runtime.bindings[TASK_ID]["job_id"]
controller = runtime.controllers[TASK_ID]
with controller.store._connect() as connection:
row = connection.execute(
"SELECT request_json FROM jobs WHERE job_id=?", (expected_job_id,)
).fetchone()
request = json.loads(row[0])
request["expected_branch_oid"] = "c" * 40
request["binding"]["expected_branch_oid"] = "c" * 40
connection.execute(
"UPDATE jobs SET request_json=? WHERE job_id=?",
(json.dumps(request, sort_keys=True, separators=(",", ":")), expected_job_id),
)
connection.commit()
with pytest.raises(RuntimeError, match="RUNTIME_GOVERNED_JOB_UNRESOLVED"):
runtime.bootstrap_writer_lease(writer_request())
assert runtime.runtimes == {}
def _commit_ready_runtime(tmp_path: Path):
runtime = make_runtime(tmp_path)
runtime.bootstrap_writer_lease(writer_request())
binding, controller, _ = runtime.runtimes[(TASK_ID, 1)]
class Workspace:
def materialize_files(self, _files):
return None
class Executor:
def __init__(self):
self.workspace = Workspace()
self.calls = 0
def execute(self, _binding, _commit, *, before_complete=None):
self.calls += 1
result = {
"job_id": binding.job_id,
"state": "COMPLETED",
"side_effect_state": "REMOTE_VERIFIED",
"receipt": {
"authority_oid": "b" * 40,
"git_remote_oid": "b" * 40,
"api_branch_oid": "b" * 40,
"commit_visible": True,
"remote_verified": True,
"force_push": False,
},
}
if before_complete is not None:
before_complete("b" * 40)
return result
executor = Executor()
runtime.runtimes[(TASK_ID, 1)] = (binding, controller, executor)
runtime.acquire_lease(writer_request())
return runtime, binding, controller, executor
def _commit_request():
return {
"task_id": TASK_ID,
"repository": "aios/aios",
"generation": 1,
"target_branch": BRANCH,
"expected_branch_oid": BASE,
"idempotency_key": "commit-cas-test-v1",
"path": "docs/cas.txt",
"content_base64": "YQ==",
"message": "test commit cas",
}
def test_commit_lease_revision_cas_conflict_is_preserved(tmp_path: Path) -> None:
runtime, _binding, controller, _executor = _commit_ready_runtime(tmp_path)
def conflict(*_args, **_kwargs):
raise JobError("LEASE_REVISION_CAS_CONFLICT")
controller.update_lease = conflict
with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
runtime.commit_files(_commit_request())
def test_commit_divergent_durable_head_fails_before_executor(tmp_path: Path) -> None:
runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
controller.update_lease(
"lease-" + binding.job_id,
expected_revision=0,
current_head_oid="c" * 40,
generation=1,
)
with pytest.raises(RuntimeError, match="LEASE_HEAD_RECONCILIATION_REQUIRED"):
runtime.commit_files(_commit_request())
assert executor.calls == 0
def test_remote_pending_never_advances_durable_head(tmp_path: Path) -> None:
runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
def pending(_binding, _commit, **_kwargs):
executor.calls += 1
return {
"job_id": binding.job_id,
"state": "RECONCILIATION_REQUIRED",
"side_effect_state": "REMOTE_PUSH_PENDING",
"receipt": {
"authority_oid": "b" * 40,
"desired_remote_oid": "b" * 40,
"remote_verified": False,
},
}
executor.execute = pending
with pytest.raises(RuntimeError, match="RECONCILIATION_REQUIRED"):
runtime.commit_files(_commit_request())
with controller.store._connect() as connection:
row = connection.execute(
"SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
("lease-" + binding.job_id,),
).fetchone()
assert tuple(row) == (BASE, 0)
def test_retry_accepts_exact_head_already_advanced_at_remote_verified_boundary(tmp_path: Path) -> None:
runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
commit_oid = "b" * 40
controller.update_lease(
"lease-" + binding.job_id,
expected_revision=0,
current_head_oid=commit_oid,
generation=1,
)
receipt = {
"authority_oid": commit_oid,
"git_remote_oid": commit_oid,
"api_branch_oid": commit_oid,
"commit_visible": True,
"remote_verified": True,
"force_push": False,
}
with controller.store._connect() as connection:
connection.execute(
"UPDATE jobs SET state='REMOTE_VERIFIED',side_effect_state='REMOTE_VERIFIED',"
"receipt_json=? WHERE job_id=?",
(json.dumps(receipt, sort_keys=True, separators=(",", ":")), binding.job_id),
)
connection.commit()
result = runtime.commit_files(_commit_request())
assert result["commit_oid"] == commit_oid
assert executor.calls == 1
with controller.store._connect() as connection:
row = connection.execute(
"SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
("lease-" + binding.job_id,),
).fetchone()
assert tuple(row) == (commit_oid, 1)
def test_completed_response_with_invalid_oid_uses_canonical_oid_error(tmp_path: Path) -> None:
runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
def invalid_oid(_binding, _commit, **_kwargs):
executor.calls += 1
return {
"job_id": binding.job_id,
"state": "COMPLETED",
"side_effect_state": "REMOTE_VERIFIED",
"receipt": {
"authority_oid": "not-an-oid",
"commit_visible": True,
"remote_verified": True,
},
}
executor.execute = invalid_oid
with pytest.raises(RuntimeError, match="GOVERNED_COMMIT_OID_INVALID"):
runtime.commit_files(_commit_request())
with controller.store._connect() as connection:
row = connection.execute(
"SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
("lease-" + binding.job_id,),
).fetchone()
assert tuple(row) == (BASE, 0)