feat(gitops): implement integration refresh provisioner

active-runtime/app/aios_autonomous_release_authorization/controller.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+"""Task-scoped continuous authorization controller for the AIOS Candidate writer."""
+from __future__ import annotations
+import base64, fcntl, hashlib, json, os, pwd, grp, re, socket, sqlite3, stat, subprocess, sys, tempfile, time, urllib.error, urllib.parse, urllib.request
+from datetime import datetime, timedelta, timezone
+from pathlib import Path, PurePosixPath, PureWindowsPath
+from typing import Any
+
+TASK='AIOS-AUTONOMOUS-RELEASE-FULL-IMPLEMENTATION-20260726'
+AUTH_ID='authz-aios-autonomous-release-20260726'
+BRANCH='patch/chatgpt-autonomous-completion-contract-20260726'
+BASE='875d41c3101093149414292067fd700dc214d4d3'
+REPO='aios/aios'
+WRITER_AUDIENCE='chatgpt-autonomous-release-writer'
+ACTOR='chatgpt-gitops-bot'
+WRITER_INSTANCE='writer-6e32f321-1240-46f8-9693-0b0631d9c77b'
+CHILD_OPERATION_MAX_USES=20
+ROOT=Path('/data/aios-gitops')
+WORKTREE=ROOT/'worktrees/chatgpt-autonomous-completion-contract-20260726'
+BARE=ROOT/'repo.git'
+TASK_ROOT=ROOT/'runtime/tasks'/TASK
+AUTH_DIR=Path('/etc/aios-gitops/task-authorizations')
+PARENT=AUTH_DIR/'authz-aios-autonomous-release-20260726.json'
+STATE_DIR=Path('/var/lib/aios-autonomous-release-authorization')
+STATE=STATE_DIR/'state.json'
+LOCK=STATE_DIR/'controller.lock'
+AUDIT=TASK_ROOT/'continuous-authorization.audit.jsonl'
+SOCKET_DIR=Path('/run/aios-autonomous-release-authorization')
+SOCKET_PATH=SOCKET_DIR/'controller.sock'
+PROJECTION=Path('/etc/aios-gitops/credentials/chatgpt-autonomous-release-writer')
+REGISTRY=Path('/var/lib/aios-chatgpt-autonomous-release-writer/token-registry.sqlite3')
+WRITER_LEDGER=Path('/var/lib/aios-chatgpt-autonomous-release-writer/operation-ledger.sqlite3')
+ASKPASS=Path('/usr/libexec/aios-autonomous-release-authorization/gitea_askpass.py')
+PATH_POLICY=Path('/etc/aios-gitops/policies-chatgpt-autonomous-release/autonomous-release-path-policy.json')
+ROUTE_BINDING=Path('/etc/aios-gitops/action-route-bindings/chatgpt-autonomous-release-20260726.json')
+OID_RE=re.compile(r'^[0-9a-f]{40}$')
+OP_RE=re.compile(r'^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$')
+TERMINAL={'COMPLETED','CANCELLED','FAILED_FINAL','REVOKED','REVOKED_IDLE_TIMEOUT','REVOKED_WALL_CLOCK','RECONCILIATION_REQUIRED'}
+
+sys.path.insert(0,'/usr/libexec/aios-chatgpt-autonomous-release-writer-fullprofile-8ac2f90f')
+from isolated_writer_token_registry import TokenRegistry
+
+def now() -> datetime: return datetime.now(timezone.utc)
+def stamp() -> str: return now().replace(microsecond=0).isoformat()
+def canonical(value: Any) -> bytes: return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode('utf-8')
+def fail(code: str, status: int=503, **details: Any) -> None: raise ControlError(code,status,details)
+class ControlError(RuntimeError):
+ def __init__(self,code:str,status:int=503,details:dict[str,Any]|None=None): self.code,self.status,self.details=code,status,details or {}; super().__init__(code)
+
+def owned_regular(path:Path, mode:int) -> os.stat_result:
+ try: info=os.lstat(path)
+ except FileNotFoundError: fail('authorization_config_missing')
+ if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_uid!=0 or stat.S_IMODE(info.st_mode)&~mode: fail('authorization_config_permissions_rejected')
+ return info
+
+def load_json(path:Path,mode:int=0o600) -> dict[str,Any]:
+ owned_regular(path,mode)
+ try: value=json.loads(path.read_text(encoding='utf-8'))
+ except Exception: fail('authorization_config_json_rejected')
+ if not isinstance(value,dict): fail('authorization_config_schema_rejected')
+ return value
+
+def atomic_json(path:Path,value:dict[str,Any],mode:int=0o600) -> None:
+ path.parent.mkdir(parents=True,exist_ok=True)
+ raw=canonical(value)+b'\n'
+ fd,tmp=tempfile.mkstemp(prefix='.new-',dir=path.parent)
+ try:
+  os.fchmod(fd,mode); os.write(fd,raw); os.fsync(fd); os.close(fd); os.chown(tmp,0,0); os.replace(tmp,path)
+  os.chmod(path,mode)
+ finally:
+  try: os.close(fd)
+  except OSError: pass
+  if os.path.exists(tmp): os.unlink(tmp)
+
+def append_audit(event:dict[str,Any]) -> None:
+ TASK_ROOT.mkdir(parents=True,exist_ok=True)
+ flags=os.O_WRONLY|os.O_CREAT|os.O_APPEND|getattr(os,'O_NOFOLLOW',0)
+ fd=os.open(AUDIT,flags,0o600)
+ try: os.write(fd,canonical({'at':stamp(),'task_id':TASK,'secrets_included':False,**event})+b'\n'); os.fsync(fd)
+ finally: os.close(fd)
+
+def parent() -> dict[str,Any]:
+ p=load_json(PARENT)
+ required={'schema_version','development_authorization_id','authorization_type','task_id','actor_identity','audience','repository','exact_branch','approved_base_oid','allowed_path_roots','allowed_exact_paths','allowed_operations','created_at','valid_until','maximum_wall_clock_seconds','idle_timeout_seconds','manual_reapproval_per_commit','maximum_concurrent_writes','path_policy_sha256','writer_instance_id'}
+ if set(p)!=required or p.get('schema_version')!=1: fail('parent_authorization_schema_rejected')
+ fixed={'development_authorization_id':AUTH_ID,'authorization_type':'TASK_SCOPED_CONTINUOUS_DEVELOPMENT','task_id':TASK,'actor_identity':ACTOR,'audience':WRITER_AUDIENCE,'repository':REPO,'exact_branch':BRANCH,'approved_base_oid':BASE,'manual_reapproval_per_commit':False,'maximum_concurrent_writes':1,'writer_instance_id':WRITER_INSTANCE}
+ if any(p.get(k)!=v for k,v in fixed.items()): fail('parent_authorization_binding_rejected')
+ if p.get('maximum_wall_clock_seconds')!=86400 or p.get('idle_timeout_seconds')!=2700: fail('parent_authorization_lifetime_rejected')
+ if not isinstance(p.get('allowed_path_roots'),list) or not isinstance(p.get('allowed_exact_paths'),list) or not isinstance(p.get('allowed_operations'),list): fail('parent_authorization_scope_rejected')
+ raw=PATH_POLICY.read_bytes()
+ if hashlib.sha256(raw).hexdigest()!=p.get('path_policy_sha256'): fail('parent_authorization_path_policy_changed')
+ return p
+
+def parse_time(v:Any) -> datetime:
+ try:
+  d=datetime.fromisoformat(str(v).replace('Z','+00:00'))
+  return d.astimezone(timezone.utc)
+ except Exception: fail('authorization_time_rejected')
+
+def components(path:str)->tuple[str,...]:
+ if not isinstance(path,str) or not path or path!=path.strip() or '\\' in path or '\x00' in path: fail('authorization_path_rejected',403)
+ pp=PurePosixPath(path); wp=PureWindowsPath(path)
+ parts=tuple(path.split('/'))
+ if pp.is_absolute() or wp.is_absolute() or bool(wp.drive) or any(x in {'','.','..'} for x in parts): fail('authorization_path_rejected',403)
+ return parts
+
+def path_allowed(p:dict[str,Any],path:str)->bool:
+ item=components(path)
+ for root in p['allowed_path_roots']:
+  r=components(str(root).rstrip('/'))
+  if item[:len(r)]==r: return True
+ return any(item==components(x) for x in p['allowed_exact_paths'])
+
+def git(*args:str,cwd:Path|None=None,env:dict[str,str]|None=None)->str:
+ e=os.environ.copy(); e.update(env or {}); e['GIT_CONFIG_COUNT']='1'; e['GIT_CONFIG_KEY_0']='safe.directory'; e['GIT_CONFIG_VALUE_0']=str(WORKTREE)
+ r=subprocess.run(['git',*args],cwd=cwd,text=True,capture_output=True,env=e)
+ if r.returncode: fail('authorization_git_probe_failed',503)
+ return r.stdout.strip()
+
+def credential_lines()->list[str]:
+ cp=Path(os.environ.get('CREDENTIALS_DIRECTORY',''))/'gitea_git_http'
+ try:
+  info=os.lstat(cp); lines=cp.read_text(encoding='utf-8').splitlines()
+ except OSError: fail('authorization_gitea_credential_unavailable')
+ if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or len(lines)!=2 or lines[0]!='gitops-bot' or not lines[1]: fail('authorization_gitea_credential_rejected')
+ return lines
+
+def remote_oid()->str:
+ credential_lines()
+ e={'GIT_ASKPASS':str(ASKPASS),'GIT_TERMINAL_PROMPT':'0'}
+ out=git('ls-remote','--heads','http://127.0.0.1:3000/aios/aios.git',f'refs/heads/{BRANCH}',cwd=WORKTREE,env=e)
+ fields=out.split()
+ if len(fields)!=2 or not OID_RE.fullmatch(fields[0]): fail('authorization_git_ls_remote_rejected')
+ return fields[0]
+
+def api_oid()->str:
+ user,password=credential_lines(); branch=urllib.parse.quote(BRANCH,safe='')
+ encoded=base64.b64encode((user+':'+password).encode()).decode()
+ request=urllib.request.Request('http://127.0.0.1:3000/api/v1/repos/aios/aios/branches/'+branch,headers={'Authorization':'Basic '+encoded,'Host':'g.cnnetmkt.top'})
+ try:
+  with urllib.request.urlopen(request,timeout=10) as response: value=json.loads(response.read())
+ except Exception: fail('authorization_gitea_api_unavailable')
+ oid=str(value.get('commit',{}).get('id',''))
+ if not OID_RE.fullmatch(oid): fail('authorization_gitea_api_oid_rejected')
+ return oid
+
+def four_way(expected:str)->None:
+ local=git('rev-parse','HEAD',cwd=WORKTREE)
+ bare=git('--git-dir',str(BARE),'rev-parse',f'refs/heads/{BRANCH}')
+ remote=remote_oid(); api=api_oid()
+ if any(x!=expected for x in (local,bare,remote,api)): fail('RECONCILIATION_REQUIRED',409,current_head_oid=local)
+ if git('status','--porcelain',cwd=WORKTREE)!='': fail('authorization_worktree_dirty',409)
+
+def state()->dict[str,Any]:
+ s=load_json(STATE)
+ required={'schema_version','development_authorization_id','task_id','status','current_head_oid','last_activity_at','writer_child','test_child','inflight','updated_at'}
+ if set(s)!=required or s.get('schema_version')!=1 or s.get('development_authorization_id')!=AUTH_ID or s.get('task_id')!=TASK or not OID_RE.fullmatch(str(s.get('current_head_oid',''))): fail('authorization_state_schema_rejected')
+ return s
+
+def save_state(s:dict[str,Any])->None: s['updated_at']=stamp(); atomic_json(STATE,s)
+
+def lock():
+ STATE_DIR.mkdir(parents=True,exist_ok=True); os.chmod(STATE_DIR,0o700)
+ f=open(LOCK,'a+b'); os.chmod(LOCK,0o600); fcntl.flock(f.fileno(),fcntl.LOCK_EX); return f
+
+def route_binding_valid(p:dict[str,Any])->None:
+ b=load_json(ROUTE_BINDING,0o640)
+ identities=b.get('action_identities')
+ if b.get('task_id')!=TASK or b.get('writer_instance_id')!=WRITER_INSTANCE or b.get('repository')!=REPO or b.get('branch')!=BRANCH or b.get('approved_base_oid')!=BASE or b.get('path_policy_sha256')!=p.get('path_policy_sha256') or not isinstance(identities,list) or len(identities)!=1 or identities[0].get('account')!=ACTOR:
+  fail('authorization_route_binding_changed',409)
+
+def task_live(p:dict[str,Any],s:dict[str,Any],*,touch:bool)->None:
+ route_binding_valid(p)
+ if s['status'] in TERMINAL: fail('authorization_task_terminal',409)
+ if now()>=parse_time(p['valid_until']):
+  s['status']='REVOKED_WALL_CLOCK'; save_state(s); cleanup(s,'wall_clock'); fail('authorization_task_terminal',409)
+ if (now()-parse_time(s['last_activity_at'])).total_seconds()>p['idle_timeout_seconds']:
+  s['status']='REVOKED_IDLE_TIMEOUT'; save_state(s); cleanup(s,'idle_timeout'); fail('authorization_task_terminal',409)
+ if touch: s['last_activity_at']=stamp(); save_state(s)
+
+def projection_token()->str|None:
+ if not PROJECTION.exists(): return None
+ try:
+  info=os.lstat(PROJECTION)
+  if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_uid!=0 or stat.S_IMODE(info.st_mode)!=0o600: fail('authorization_projection_rejected')
+  token=PROJECTION.read_text(encoding='utf-8').strip()
+  return token if len(token.encode())>=32 else None
+ except OSError: fail('authorization_projection_rejected')
+
+def write_projection(token:str)->None:
+ PROJECTION.parent.mkdir(parents=True,exist_ok=True)
+ fd,tmp=tempfile.mkstemp(prefix='.writer-',dir=PROJECTION.parent)
+ try:
+  os.fchmod(fd,0o600); os.write(fd,(token+'\n').encode()); os.fsync(fd); os.close(fd); os.chown(tmp,0,0); os.replace(tmp,PROJECTION); os.chmod(PROJECTION,0o600)
+ finally:
+  try: os.close(fd)
+  except OSError: pass
+  if os.path.exists(tmp): os.unlink(tmp)
+
+def post_revoke_401(token:str)->None:
+ req=urllib.request.Request('http://127.0.0.1:19893/api/runtime-status',headers={'Authorization':'Bearer '+token})
+ try: urllib.request.urlopen(req,timeout=5)
+ except urllib.error.HTTPError as e:
+  if e.code==401:return
+ except Exception: pass
+ fail('authorization_old_credential_revocation_unverified')
+
+def rotate_child(p:dict[str,Any],s:dict[str,Any],reason:str)->str:
+ old=projection_token(); registry=TokenRegistry(REGISTRY)
+ old_record=registry.find(old) if old else None
+ if old_record and (old_record.get('task_id')!=TASK or old_record.get('development_authorization_id')!=AUTH_ID):
+  if parse_time(old_record.get('expires_at'))>=now(): fail('authorization_existing_projection_binding_rejected')
+  if not registry.revoke(str(old_record['token_id'])): fail('authorization_historical_child_revoke_failed')
+  post_revoke_401(old)
+  append_audit({'event':'historical_expired_child_revoked','old_credential_http_status':401})
+  old_record=None
+ issued=now().replace(microsecond=0); expires=issued+timedelta(minutes=20)
+ prefixes=[str(x).rstrip('/')+'/' for x in p['allowed_path_roots']]
+ grant={'token_id':'continuous-child-'+issued.strftime('%Y%m%dT%H%M%SZ')+'-'+os.urandom(3).hex(),'credential_id':'credential-'+issued.strftime('%Y%m%dT%H%M%SZ')+'-'+os.urandom(3).hex(),'development_authorization_id':AUTH_ID,'audience':WRITER_AUDIENCE,'task_id':TASK,'actor_identity':ACTOR,'writer_instance_id':WRITER_INSTANCE,'repository_identity':REPO,'source_cidrs':['127.0.0.1/32'],'allowed_branch':BRANCH,'current_expected_head_oid':s['current_head_oid'],'approved_base_oid':BASE,'allowed_operations':p['allowed_operations'],'allowed_paths':p['allowed_exact_paths'],'allowed_path_prefixes':prefixes,'not_before':issued.isoformat(),'expires_at':expires.isoformat(),'renewable':False,'max_uses':{operation:CHILD_OPERATION_MAX_USES for operation in p['allowed_operations']},'issued_by':'task_authorization_controller'}
+ token,record=registry.issue(grant)
+ try:
+  write_projection(token)
+  if old_record:
+   if not registry.revoke(str(old_record['token_id'])): fail('authorization_old_child_revoke_failed')
+   post_revoke_401(old)
+ except Exception:
+  registry.revoke(str(record['token_id']))
+  try: PROJECTION.unlink()
+  except FileNotFoundError: pass
+  raise
+ s['writer_child']={'credential_id':record['credential_id'],'issued_at':record['not_before'],'expires_at':record['expires_at'],'current_expected_head_oid':s['current_head_oid'],'rotation_reason':reason}
+ save_state(s); append_audit({'event':'child_grant_rotated','reason':reason,'current_head_oid':s['current_head_oid'],'old_grant_revoked':bool(old_record),'old_credential_http_status':401 if old_record else None})
+ return token
+
+def child_operation_quota_exhausted(rec:dict[str,Any],registry:TokenRegistry)->bool:
+ token_id=str(rec.get('token_id',''))
+ allowed=rec.get('allowed_operations')
+ limits=rec.get('max_uses')
+ if not token_id or not isinstance(allowed,list) or not isinstance(limits,dict): fail('authorization_child_quota_schema_rejected')
+ for operation in allowed:
+  maximum=limits.get(operation)
+  if not isinstance(operation,str) or not isinstance(maximum,int) or maximum<1: fail('authorization_child_quota_schema_rejected')
+  if registry.usage(token_id,operation)>=maximum: return True
+ return False
+
+def ensure_child(p:dict[str,Any],s:dict[str,Any],reason:str)->str:
+ token=projection_token(); meta=s.get('writer_child') or {}
+ expired=True
+ try: expired=(parse_time(meta.get('expires_at'))-now()).total_seconds()<=300
+ except Exception: pass
+ if not token or expired or meta.get('current_expected_head_oid')!=s['current_head_oid']:
+  return rotate_child(p,s,reason)
+ registry=TokenRegistry(REGISTRY)
+ rec=registry.find(token)
+ if not rec or rec.get('revoked') or rec.get('development_authorization_id')!=AUTH_ID or rec.get('current_expected_head_oid')!=s['current_head_oid']:
+  return rotate_child(p,s,reason)
+ if child_operation_quota_exhausted(rec,registry):
+  return rotate_child(p,s,'operation_quota_exhausted')
+ return token
+
+def ledger_verified(operation_id:str,commit_oid:str)->None:
+ with sqlite3.connect(WRITER_LEDGER) as db:
+  row=db.execute('select operation_id,branch,commit_oid,state,verification_status from writer_operations where operation_id=?',(operation_id,)).fetchone()
+ if not row or row[1]!=BRANCH or row[2]!=commit_oid or row[3]!='COMPLETED' or row[4]!='bare+gitea': fail('authorization_writer_ledger_rejected',409)
+
+def reconcile(p:dict[str,Any],s:dict[str,Any],previous:str,new:str,operation_id:str)->dict[str,Any]:
+ inflight=s.get('inflight')
+ if not isinstance(inflight,dict) or inflight.get('previous_head_oid')!=previous or inflight.get('operation_id')!=operation_id: fail('authorization_inflight_rejected',409)
+ if previous!=s['current_head_oid'] or not OID_RE.fullmatch(new): fail('authorization_head_transition_rejected',409)
+ ledger_verified(operation_id,new); four_way(new)
+ s['current_head_oid']=new; s['inflight']=None; s['last_activity_at']=stamp(); save_state(s)
+ ensure_child(p,s,'head_advance')
+ append_audit({'event':'authorization_head_advanced','previous_head_oid':previous,'new_commit_oid':new,'four_way_ref_status':True,'worktree_clean':True})
+ return {'authorization_current_head_oid':new,'automatic_head_advance':True,'four_way_ref_status':True}
+
+def prepare(p:dict[str,Any],s:dict[str,Any],data:dict[str,Any])->dict[str,Any]:
+ required={'expected_current_head_oid','operation_id','idempotency_key','exact_path','content_digest','commit_message_digest','branch','path','message'}
+ if set(data)!=required: fail('authorization_request_schema_rejected',403)
+ if data['branch']!=BRANCH or data['path']!=data['exact_path'] or not path_allowed(p,str(data['path'])): fail('authorization_path_rejected',403)
+ if data['expected_current_head_oid']!=s['current_head_oid'] or not OID_RE.fullmatch(str(data['expected_current_head_oid'])): fail('BRANCH_HEAD_CHANGED',409,current_head_oid=s['current_head_oid'])
+ if not OP_RE.fullmatch(str(data['operation_id'])) or not OP_RE.fullmatch(str(data['idempotency_key'])): fail('authorization_operation_id_rejected',403)
+ if not all(isinstance(data[k],str) and re.fullmatch(r'[0-9a-f]{64}',data[k]) for k in ('content_digest','commit_message_digest')): fail('authorization_digest_rejected',403)
+ if hashlib.sha256(data['message'].encode()).hexdigest()!=data['commit_message_digest']: fail('authorization_commit_message_digest_rejected',403)
+ inflight=s.get('inflight')
+ if inflight:
+  if inflight.get('operation_id')==data['operation_id'] and inflight.get('idempotency_key')==data['idempotency_key']: return {'expected_head_oid':s['current_head_oid'],'idempotency_key':data['idempotency_key']}
+  fail('BRANCH_HEAD_CHANGED',409,current_head_oid=s['current_head_oid'])
+ four_way(s['current_head_oid'])
+ s['inflight']={'operation_id':data['operation_id'],'idempotency_key':data['idempotency_key'],'previous_head_oid':s['current_head_oid'],'path':data['path'],'created_at':stamp()}; s['last_activity_at']=stamp(); save_state(s)
+ append_audit({'event':'write_prepared','operation_id':data['operation_id'],'path':data['path'],'expected_head_oid':s['current_head_oid']})
+ return {'expected_head_oid':s['current_head_oid'],'idempotency_key':data['idempotency_key']}
+
+def test_authorize(p:dict[str,Any],s:dict[str,Any],operation:str)->dict[str,Any]:
+ allowed={'pytest_dormant','pytest_autonomous_release','compileall_autonomous_release','git_diff_check'}
+ if operation not in allowed: fail('candidate_test_executor_operation_rejected',403)
+ four_way(s['current_head_oid'])
+ child=s.get('test_child') or {}; refresh=True
+ try: refresh=(parse_time(child.get('expires_at'))-now()).total_seconds()<=300
+ except Exception: pass
+ if refresh:
+  issued=now().replace(microsecond=0); child={'test_authorization_id':'test-authz-aios-autonomous-release-20260726','grant_id':'test-child-'+issued.strftime('%Y%m%dT%H%M%SZ'),'issued_at':issued.isoformat(),'expires_at':(issued+timedelta(minutes=20)).isoformat()}; s['test_child']=child
+ s['last_activity_at']=stamp(); save_state(s); append_audit({'event':'test_authorized','operation':operation})
+ return {'test_authorization_id':child['test_authorization_id'],'operation':operation,'expires_at':child['expires_at']}
+
+def cleanup(s:dict[str,Any],reason:str)->None:
+ old=projection_token()
+ if old:
+  reg=TokenRegistry(REGISTRY); rec=reg.find(old)
+  if rec: reg.revoke(str(rec['token_id']))
+  try: PROJECTION.unlink()
+  except FileNotFoundError: pass
+  try: post_revoke_401(old)
+  except ControlError: pass
+ for unit in ('aios-chatgpt-autonomous-release-route.service','aios-chatgpt-autonomous-release-writer.service'):
+  subprocess.run(['/usr/bin/systemctl','stop',unit],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=20)
+ subprocess.run(['/usr/bin/systemctl','stop','aios-autonomous-release-dormant-test-executor@*.service'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=20)
+ try: ROUTE_BINDING.unlink()
+ except FileNotFoundError: pass
+ try: SOCKET_PATH.unlink()
+ except FileNotFoundError: pass
+ append_audit({'event':'authorization_cleanup','reason':reason,'projection_removed':not PROJECTION.exists(),'route_overlay_removed':not ROUTE_BINDING.exists(),'task_services_stop_requested':True})
+
+def request(data:dict[str,Any])->dict[str,Any]:
+ op=data.get('op'); f=lock()
+ try:
+  p=parent(); s=state(); task_live(p,s,touch=False)
+  if op=='describe_authorization':
+   four_way(s['current_head_oid'])
+   return {'development_authorization_id':AUTH_ID,'task_id':TASK,'actor_identity':ACTOR,'audience':WRITER_AUDIENCE,'repository':REPO,'branch':BRANCH,'approved_base_oid':BASE,'authorization_status':s['status'],'current_head_oid':s['current_head_oid']}
+  if op=='get_writer_token':
+   four_way(s['current_head_oid']); token=ensure_child(p,s,'scheduled_or_requested_rotation'); return {'token':token,'current_head_oid':s['current_head_oid']}
+  if op=='prepare_write':
+   ensure_child(p,s,'pre_write_quota_check')
+   return prepare(p,s,data.get('request') if isinstance(data.get('request'),dict) else {})
+  if op=='reconcile_commit':
+   r=data.get('result') if isinstance(data.get('result'),dict) else {}; return reconcile(p,s,str(r.get('previous_head_oid','')),str(r.get('new_commit_oid','')),str(r.get('operation_id','')))
+  if op=='fail_write':
+   r=data.get('result') if isinstance(data.get('result'),dict) else {}; inflight=s.get('inflight')
+   if isinstance(inflight,dict) and inflight.get('operation_id')==r.get('operation_id'): s['inflight']=None; save_state(s)
+   return {'cleared':True}
+  if op=='authorize_test': return test_authorize(p,s,str(data.get('operation','')))
+  fail('authorization_operation_rejected',403)
+ finally:
+  fcntl.flock(f.fileno(),fcntl.LOCK_UN); f.close()
+
+def peer_allowed(conn:socket.socket)->bool:
+ try:
+  import struct
+  _pid,uid,gid=struct.unpack('3i',conn.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))
+  return uid==pwd.getpwnam('aiman').pw_uid and gid==grp.getgrnam('gitops').gr_gid
+ except Exception: return False
+
+def serve()->None:
+ SOCKET_DIR.mkdir(parents=True,exist_ok=True); os.chown(SOCKET_DIR,0,grp.getgrnam('gitops').gr_gid); os.chmod(SOCKET_DIR,0o750)
+ try: SOCKET_PATH.unlink()
+ except FileNotFoundError: pass
+ with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as server:
+  server.bind(str(SOCKET_PATH)); os.chown(SOCKET_PATH,0,grp.getgrnam('gitops').gr_gid); os.chmod(SOCKET_PATH,0o660); server.listen(32); server.settimeout(1.0)
+  last=0.0
+  while True:
+   if time.monotonic()-last>30:
+    last=time.monotonic()
+    try:
+     f=lock(); p=parent(); s=state(); task_live(p,s,touch=False); four_way(s['current_head_oid']); ensure_child(p,s,'scheduled_rotation'); fcntl.flock(f.fileno(),fcntl.LOCK_UN); f.close()
+    except ControlError as e:
+     append_audit({'event':'scheduled_check_blocked','error':e.code})
+     if e.code=='authorization_task_terminal': return
+    except Exception: append_audit({'event':'scheduled_check_blocked','error':'authorization_controller_internal_error'})
+   try: conn,_=server.accept()
+   except socket.timeout: continue
+   with conn:
+    if not peer_allowed(conn): conn.sendall(canonical({'ok':False,'status':403,'error':'authorization_peer_rejected'})+b'\n'); continue
+    try:
+     raw=conn.recv(131072); data=json.loads(raw.decode()); result=request(data); conn.sendall(canonical({'ok':True,'result':result})+b'\n')
+    except ControlError as e: conn.sendall(canonical({'ok':False,'status':e.status,'error':e.code,**e.details})+b'\n')
+    except Exception as exc:
+     append_audit({'event':'controller_internal_error','error_type':type(exc).__name__})
+     conn.sendall(canonical({'ok':False,'status':503,'error':'authorization_controller_internal_error'})+b'\n')
+if __name__=='__main__': serve()

active-runtime/app/aios_chatgpt_action_task_router/action_task_router.py

@@ -0,0 +1,567 @@
+#!/usr/bin/env python3
+"""Task-bound bridge from the legacy ChatGPT Action to an isolated Writer V2."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import re
+import sqlite3
+import socket
+import stat
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path, PurePosixPath, PureWindowsPath
+from typing import Any
+
+
+CONFIG_PATH = Path(os.environ["AIOS_ACTION_ROUTE_BINDING"])
+POLICY_PATH = Path(os.environ["AIOS_ACTION_ROUTE_PATH_POLICY"])
+KNOWN_IDENTITIES_PATH = Path(os.environ["AIOS_ACTION_ROUTE_KNOWN_IDENTITIES"])
+STATE_PATH = Path(os.environ["AIOS_ACTION_ROUTE_STATE"])
+AUDIT_PATH = Path(os.environ["AIOS_ACTION_ROUTE_AUDIT"])
+HOST = os.environ.get("AIOS_ACTION_ROUTE_HOST", "127.0.0.1")
+PORT = int(os.environ.get("AIOS_ACTION_ROUTE_PORT", "19894"))
+EXPECTED_TASK_ID = os.environ["AIOS_ACTION_ROUTE_TASK_ID"]
+EXPECTED_ROUTE_HEADER = os.environ["AIOS_ACTION_ROUTE_HEADER"]
+AUTHORIZATION_SOCKET = Path(os.environ["AIOS_CONTINUOUS_AUTHORIZATION_SOCKET"])
+OID_RE = re.compile(r"[0-9a-f]{40}")
+SECRET_RE = re.compile(
+    r"(?i)(api[_-]?key|secret[_-]?access[_-]?key|secret|password|token)"
+    r"\s*[:=]\s*[\"']?([A-Za-z0-9_./+=:-]{16,})"
+)
+SECRET_VALUE_RE = re.compile(
+    r"(sk-[A-Za-z0-9_-]{20,}|HPUA[A-Z0-9]{10,}|SDC[A-Za-z0-9]{20,}|"
+    r"-----BEGIN (?:OPENSSH |RSA |EC )?PRIVATE KEY-----)"
+)
+ALLOWED_REQUEST_FIELDS = frozenset({"base","branch","content","create_branch","message","path","task_id","development_authorization_id","actor_identity","audience","approved_base_oid","expected_current_head_oid","operation_id","idempotency_key","exact_path","content_digest","commit_message_digest"})
+
+
+class RouteError(RuntimeError):
+    def __init__(self, code: str, status: int = 400, **details: Any):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+        self.details = details
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def canonical_json(value: Any) -> bytes:
+    return json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def load_root_managed_json(
+    path: Path, *, expected_mode: int = 0o640
+) -> tuple[dict[str, Any], bytes]:
+    info = os.lstat(path)
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise RouteError("writer_route_config_type_rejected", 503)
+    if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & ~expected_mode:
+        raise RouteError("writer_route_config_permissions_rejected", 503)
+    raw = path.read_bytes()
+    try:
+        value = json.loads(raw)
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise RouteError("writer_route_config_json_rejected", 503) from None
+    if not isinstance(value, dict):
+        raise RouteError("writer_route_config_object_required", 503)
+    return value, raw
+
+
+def load_binding() -> dict[str, Any]:
+    binding, _ = load_root_managed_json(CONFIG_PATH)
+    required = {
+        "schema_version",
+        "task_id",
+        "writer_instance_id",
+        "repository",
+        "branch",
+        "approved_base_oid",
+        "writer_url",
+        "path_policy_sha256",
+        "action_identities",
+    }
+    if set(binding) != required or binding.get("schema_version") != 1:
+        raise RouteError("writer_route_binding_schema_rejected", 503)
+    if binding["task_id"] != EXPECTED_TASK_ID:
+        raise RouteError("writer_route_task_binding_rejected", 503)
+    if binding["repository"] != "aios/aios":
+        raise RouteError("writer_route_repository_rejected", 503)
+    if not OID_RE.fullmatch(str(binding["approved_base_oid"])):
+        raise RouteError("writer_route_base_oid_rejected", 503)
+    if binding["writer_url"] != "http://127.0.0.1:19893":
+        raise RouteError("writer_route_upstream_rejected", 503)
+    identities = binding.get("action_identities")
+    if not isinstance(identities, list) or len(identities) != 1:
+        raise RouteError("writer_route_action_identity_rejected", 503)
+    identity = identities[0]
+    if (
+        not isinstance(identity, dict)
+        or set(identity) != {"account", "token_sha256"}
+        or identity.get("account") != "chatgpt-gitops-bot"
+        or not re.fullmatch(
+            r"[0-9a-f]{64}", str(identity.get("token_sha256", ""))
+        )
+    ):
+        raise RouteError("writer_route_action_identity_rejected", 503)
+    return binding
+
+
+def allowed_path_scope(binding: dict[str, Any]) -> tuple[set[str], tuple[str, ...]]:
+    policy, raw = load_root_managed_json(POLICY_PATH)
+    actual = hashlib.sha256(raw).hexdigest()
+    if actual != binding["path_policy_sha256"]:
+        raise RouteError("writer_route_path_policy_digest_mismatch", 503)
+    if policy.get("allowed_branch") != binding["branch"]:
+        raise RouteError("writer_route_path_policy_branch_mismatch", 503)
+    operation = (policy.get("allowed_operations") or {}).get("commit_file.atomic")
+    paths = operation.get("exact_paths") if isinstance(operation, dict) else None
+    prefixes = operation.get("prefix_paths") if isinstance(operation, dict) else None
+    if not isinstance(paths, list) or not paths or not all(
+        isinstance(x, str) for x in paths
+    ):
+        raise RouteError("writer_route_path_policy_rejected", 503)
+    if not isinstance(prefixes, list) or not prefixes or not all(
+        isinstance(x, str) and x.endswith("/") for x in prefixes
+    ):
+        raise RouteError("writer_route_path_policy_rejected", 503)
+    if any(not valid_relative_path(x[:-1]) for x in prefixes):
+        raise RouteError("writer_route_path_policy_rejected", 503)
+    if any(not valid_relative_path(x) for x in paths):
+        raise RouteError("writer_route_path_policy_rejected", 503)
+    return set(paths), tuple(prefixes)
+
+
+def valid_relative_path(path: str) -> bool:
+    if not isinstance(path, str) or not path or path != path.strip():
+        return False
+    if "\x00" in path or "\\" in path:
+        return False
+    try:
+        decoded = urllib.parse.unquote(path, errors="strict")
+    except (UnicodeDecodeError, ValueError):
+        return False
+    if decoded != path:
+        return False
+    posix = PurePosixPath(path)
+    windows = PureWindowsPath(path)
+    return not (
+        posix.is_absolute()
+        or windows.is_absolute()
+        or bool(windows.drive)
+        or any(part in {"", ".", ".."} for part in path.split("/"))
+    )
+
+
+def path_allowed(binding: dict[str, Any], path: str) -> bool:
+    if not valid_relative_path(path):
+        return False
+    exact_paths, prefixes = allowed_path_scope(binding)
+    return path in exact_paths or any(path.startswith(prefix) for prefix in prefixes)
+
+
+def bearer(headers: Any) -> str:
+    value = headers.get("Authorization", "")
+    if not value.startswith("Bearer "):
+        raise RouteError("unauthorized", 401)
+    token = value[7:].strip()
+    if not token:
+        raise RouteError("unauthorized", 401)
+    return token
+
+
+def authorize_action(headers: Any, binding: dict[str, Any]) -> str:
+    digest = hashlib.sha256(bearer(headers).encode("utf-8")).hexdigest()
+    identity = binding["action_identities"][0]
+    if not hmac.compare_digest(identity["token_sha256"], digest):
+        known, _ = load_root_managed_json(KNOWN_IDENTITIES_PATH)
+        identities = known.get("identities")
+        if known.get("schema_version") != 1 or not isinstance(identities, list):
+            raise RouteError("writer_route_identity_catalog_rejected", 503)
+        observed = "unregistered"
+        allowed_accounts = {
+            "chatgpt-gitops-bot",
+            "gitops-write-bot",
+            "codex-gitops-bot",
+        }
+        for candidate in identities:
+            if not isinstance(candidate, dict) or set(candidate) != {
+                "account",
+                "token_sha256",
+            }:
+                raise RouteError("writer_route_identity_catalog_rejected", 503)
+            account = str(candidate.get("account", ""))
+            token_sha256 = str(candidate.get("token_sha256", ""))
+            if account not in allowed_accounts or not re.fullmatch(
+                r"[0-9a-f]{64}", token_sha256
+            ):
+                raise RouteError("writer_route_identity_catalog_rejected", 503)
+            if hmac.compare_digest(token_sha256, digest):
+                observed = account
+        raise RouteError(
+            "writer_route_action_identity_rejected",
+            403,
+            accepted_action_identity=str(identity["account"]),
+            observed_action_identity=observed,
+            identity_alias_accepted=False,
+        )
+    return str(identity["account"])
+
+
+def authorization_call(operation: str, **payload: Any) -> dict[str, Any]:
+    request = {"op": operation, **payload}
+    try:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
+            client.settimeout(15)
+            client.connect(str(AUTHORIZATION_SOCKET))
+            client.sendall(canonical_json(request))
+            raw = client.recv(131072)
+        response = json.loads(raw.decode("utf-8"))
+    except (OSError, TimeoutError, UnicodeDecodeError, json.JSONDecodeError):
+        raise RouteError("authorization_controller_unavailable", 503) from None
+    if not isinstance(response, dict) or not response.get("ok"):
+        if not isinstance(response, dict):
+            raise RouteError("authorization_controller_response_rejected", 503)
+        details = {k: v for k, v in response.items() if k not in {"ok", "status", "error"}}
+        raise RouteError(str(response.get("error") or "authorization_controller_rejected"), int(response.get("status") or 503), **details)
+    result = response.get("result")
+    if not isinstance(result, dict):
+        raise RouteError("authorization_controller_response_rejected", 503)
+    return result
+
+
+def writer_token() -> str:
+    value = authorization_call("get_writer_token").get("token")
+    if not isinstance(value, str) or len(value.encode("utf-8")) < 32:
+        raise RouteError("writer_route_credential_rejected", 503)
+    return value
+
+
+def writer_call(
+    binding: dict[str, Any],
+    method: str,
+    path: str,
+    payload: dict[str, Any] | None = None,
+) -> tuple[int, dict[str, Any]]:
+    data = canonical_json(payload) if payload is not None else None
+    headers = {"Authorization": "Bearer " + writer_token()}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(
+        binding["writer_url"] + path, data=data, method=method, headers=headers
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=90) as response:
+            body = json.loads(response.read())
+            return response.status, body
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read())
+        except Exception:
+            body = {"error": "writer_route_upstream_non_json_error"}
+        return exc.code, body
+    except (OSError, TimeoutError):
+        raise RouteError("writer_route_upstream_unavailable", 503) from None
+
+
+def validate_request(data: Any, binding: dict[str, Any]) -> dict[str, Any]:
+    if not isinstance(data, dict):
+        raise RouteError("writer_route_request_object_required")
+    if not {"branch", "path", "content"}.issubset(data):
+        raise RouteError("writer_route_request_incomplete")
+    if not set(data).issubset(ALLOWED_REQUEST_FIELDS):
+        raise RouteError("writer_route_request_fields_rejected")
+    if data["branch"] != binding["branch"]:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    if data.get("create_branch") not in (None, False):
+        raise RouteError("writer_route_request_binding_failed", 403)
+    if data.get("base", "main") not in {"main", binding["approved_base_oid"]}:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    path = data["path"]
+    if not isinstance(path, str) or not path_allowed(binding, path):
+        raise RouteError("writer_route_request_binding_failed", 403)
+    content = data["content"]
+    if not isinstance(content, str) or "\x00" in content or "\r" in content:
+        raise RouteError("writer_route_content_rejected")
+    if len(content.encode("utf-8")) > 1_048_576:
+        raise RouteError("writer_route_content_too_large", 413)
+    if SECRET_RE.search(content) or SECRET_VALUE_RE.search(content):
+        raise RouteError("writer_route_secret_content_rejected", 403)
+    message = data.get("message") or f"chore(chatgpt): update {path}"
+    if not isinstance(message, str) or not 1 <= len(message) <= 200 or "\x00" in message or "\n" in message or "\r" in message:
+        raise RouteError("writer_route_message_rejected")
+    return {"branch": binding["branch"], "path": path, "content": content, "message": message}
+
+
+def complete_request(data: dict[str, Any], normalized: dict[str, Any], binding: dict[str, Any], account: str) -> dict[str, Any]:
+    authorization = authorization_call("describe_authorization")
+    expected = {
+        "task_id": binding["task_id"],
+        "development_authorization_id": authorization.get("development_authorization_id"),
+        "actor_identity": account,
+        "audience": authorization.get("audience"),
+        "branch": binding["branch"],
+        "approved_base_oid": binding["approved_base_oid"],
+    }
+    required = {
+        "task_id": EXPECTED_TASK_ID,
+        "development_authorization_id": "authz-aios-autonomous-release-20260726",
+        "actor_identity": "chatgpt-gitops-bot",
+        "audience": "chatgpt-autonomous-release-writer",
+        "branch": binding["branch"],
+        "approved_base_oid": binding["approved_base_oid"],
+    }
+    if expected != required or authorization.get("authorization_status") != "ACTIVE" or authorization.get("current_head_oid") is None:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    for key, value in required.items():
+        if key in data and data[key] != value:
+            raise RouteError("writer_route_request_binding_failed", 403)
+    if "exact_path" in data and data["exact_path"] != normalized["path"]:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    current_head = str(authorization["current_head_oid"])
+    if not OID_RE.fullmatch(current_head):
+        raise RouteError("authorization_current_head_rejected", 503)
+    if "expected_current_head_oid" in data and data["expected_current_head_oid"] != current_head:
+        raise RouteError("BRANCH_HEAD_CHANGED", 409, current_head_oid=current_head)
+    content_digest = hashlib.sha256(normalized["content"].encode("utf-8")).hexdigest()
+    message_digest = hashlib.sha256(normalized["message"].encode("utf-8")).hexdigest()
+    if "content_digest" in data and data["content_digest"] != content_digest:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    if "commit_message_digest" in data and data["commit_message_digest"] != message_digest:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    seed = {**required, "expected_current_head_oid": current_head, "path": normalized["path"], "content_digest": content_digest, "commit_message_digest": message_digest}
+    request_digest = hashlib.sha256(canonical_json(seed)).hexdigest()
+    idempotency_key = "action-" + request_digest
+    operation_id = hashlib.sha256((EXPECTED_TASK_ID + "\0" + idempotency_key).encode("utf-8")).hexdigest()
+    if "idempotency_key" in data and data["idempotency_key"] != idempotency_key:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    if "operation_id" in data and data["operation_id"] != operation_id:
+        raise RouteError("writer_route_request_binding_failed", 403)
+    return {**normalized, **required, "expected_current_head_oid": current_head, "operation_id": operation_id, "idempotency_key": idempotency_key, "exact_path": normalized["path"], "content_digest": content_digest, "commit_message_digest": message_digest, "request_digest": request_digest}
+
+
+def current_writer_head(binding: dict[str, Any]) -> str:
+    status, body = writer_call(binding, "GET", "/api/runtime-status")
+    if status != 200:
+        raise RouteError(
+            "writer_route_runtime_status_failed", 503, upstream_status=status
+        )
+    required = {
+        "service": "chatgpt-autonomous-release-writer",
+        "writes_enabled": True,
+        "allowed_branch": binding["branch"],
+        "approved_base_oid": binding["approved_base_oid"],
+        "path_policy_sha256": binding["path_policy_sha256"],
+    }
+    if any(body.get(key) != value for key, value in required.items()):
+        raise RouteError("writer_route_runtime_binding_mismatch", 503)
+    head = str(body.get("workspace_head_oid", ""))
+    if not OID_RE.fullmatch(head):
+        raise RouteError("writer_route_runtime_head_rejected", 503)
+    return head
+
+
+def operation_binding(
+    request_digest: str, binding: dict[str, Any]
+) -> tuple[str, str]:
+    STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with sqlite3.connect(STATE_PATH, timeout=30, isolation_level=None) as db:
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        db.execute(
+            """
+            CREATE TABLE IF NOT EXISTS action_operations (
+                request_digest TEXT PRIMARY KEY,
+                expected_head_oid TEXT NOT NULL,
+                idempotency_key TEXT NOT NULL UNIQUE,
+                created_at TEXT NOT NULL
+            )
+            """
+        )
+        row = db.execute(
+            "SELECT expected_head_oid,idempotency_key FROM action_operations "
+            "WHERE request_digest=?",
+            (request_digest,),
+        ).fetchone()
+        if row:
+            return str(row[0]), str(row[1])
+    head = current_writer_head(binding)
+    key = "action-" + request_digest
+    with sqlite3.connect(STATE_PATH, timeout=30, isolation_level=None) as db:
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        db.execute("BEGIN IMMEDIATE")
+        db.execute(
+            "INSERT OR IGNORE INTO action_operations"
+            "(request_digest,expected_head_oid,idempotency_key,created_at) "
+            "VALUES(?,?,?,?)",
+            (request_digest, head, key, utc_now()),
+        )
+        row = db.execute(
+            "SELECT expected_head_oid,idempotency_key FROM action_operations "
+            "WHERE request_digest=?",
+            (request_digest,),
+        ).fetchone()
+        db.execute("COMMIT")
+    return str(row[0]), str(row[1])
+
+
+def append_audit(event: dict[str, Any]) -> None:
+    AUDIT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND | getattr(os, "O_NOFOLLOW", 0)
+    fd = os.open(AUDIT_PATH, flags, 0o600)
+    try:
+        os.write(fd, canonical_json(event) + b"\n")
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "AIOSActionTaskRouter/1"
+
+    def log_message(self, _format: str, *_args: Any) -> None:
+        return
+
+    def send_json(self, status: int, value: dict[str, Any]) -> None:
+        body = json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self) -> None:
+        path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+        if path != "/health":
+            return self.send_json(404, {"error": "not_found"})
+        try:
+            binding = load_binding()
+            self.send_json(
+                200,
+                {
+                    "status": "ok",
+                    "service": "aios-action-task-router",
+                    "task_id": binding["task_id"],
+                    "writer_instance_id": binding["writer_instance_id"],
+                    "branch": binding["branch"],
+                },
+            )
+        except RouteError as exc:
+            self.send_json(exc.status, {"error": exc.code, **exc.details})
+
+    def do_POST(self) -> None:
+        path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+        if path != "/api/file":
+            return self.send_json(404, {"error": "not_found"})
+        if self.client_address[0] not in {"127.0.0.1", "::1"}:
+            return self.send_json(403, {"error": "writer_route_source_rejected"})
+        if self.headers.get("X-AIOS-Route-Task") != EXPECTED_ROUTE_HEADER:
+            return self.send_json(403, {"error": "writer_route_header_rejected"})
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+            if not 0 < length <= 1_100_000:
+                raise RouteError("writer_route_content_length_rejected", 413)
+            try:
+                data = json.loads(self.rfile.read(length).decode("utf-8"))
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                raise RouteError("writer_route_request_json_rejected") from None
+            binding = load_binding()
+            account = authorize_action(self.headers, binding)
+            normalized = validate_request(data, binding)
+            normalized = complete_request(data, normalized, binding, account)
+            prepared = authorization_call("prepare_write", request={key: normalized[key] for key in ("expected_current_head_oid", "operation_id", "idempotency_key", "exact_path", "content_digest", "commit_message_digest", "branch", "path", "message")})
+            expected_head = str(prepared["expected_head_oid"])
+            writer_payload = {"branch": normalized["branch"], "path": normalized["path"], "content": normalized["content"], "message": normalized["message"], "expected_head_oid": expected_head, "idempotency_key": str(prepared["idempotency_key"])}
+            request_digest = hashlib.sha256(canonical_json({"task_id": binding["task_id"], "operation_id": normalized["operation_id"], "idempotency_key": normalized["idempotency_key"], "content_digest": normalized["content_digest"], "commit_message_digest": normalized["commit_message_digest"]})).hexdigest()
+            status, result = writer_call(
+                binding, "POST", "/api/commit-file", writer_payload
+            )
+            audit = {
+                "at": utc_now(),
+                "task_id": binding["task_id"],
+                "writer_instance_id": binding["writer_instance_id"],
+                "account": account,
+                "branch": normalized["branch"],
+                "path": normalized["path"],
+                "request_digest": request_digest,
+                "expected_head_oid": expected_head,
+                "status": status,
+                "operation_id": result.get("operation_id"),
+                "commit_oid": result.get("commit_oid"),
+                "error": result.get("error"),
+                "secrets_included": False,
+            }
+            append_audit(audit)
+            if status != 200:
+                try:
+                    authorization_call("fail_write", result={"operation_id": normalized["operation_id"]})
+                except RouteError:
+                    pass
+                return self.send_json(status, result)
+            advanced = authorization_call("reconcile_commit", result={"previous_head_oid": expected_head, "new_commit_oid": result.get("commit_oid"), "operation_id": result.get("operation_id")})
+            return self.send_json(
+                200,
+                {
+                    "status": "committed",
+                    "branch": normalized["branch"],
+                    "path": normalized["path"],
+                    "commit": {
+                        "status": "committed",
+                        "branch": normalized["branch"],
+                        "rev": result.get("commit_oid"),
+                        "previous_head_oid": expected_head,
+                        "new_tree_oid": result.get("new_tree_oid"),
+                        "push_status": result.get("push_status"),
+                        "four_way_ref_status": advanced.get("four_way_ref_status") is True,
+                        "sync": {"status": "verified", "target": "bare+git-ls-remote+gitea-api"},
+                    },
+                    "writer_instance_id": binding["writer_instance_id"],
+                    "operation_id": result.get("operation_id"),
+                    "idempotent": result.get("idempotent"),
+                    "secrets_included": False,
+                },
+            )
+        except RouteError as exc:
+            if exc.code == "writer_route_action_identity_rejected":
+                try:
+                    append_audit(
+                        {
+                            "at": utc_now(),
+                            "task_id": EXPECTED_TASK_ID,
+                            "event": "action_identity_rejected",
+                            "accepted_action_identity": exc.details.get(
+                                "accepted_action_identity"
+                            ),
+                            "observed_action_identity": exc.details.get(
+                                "observed_action_identity"
+                            ),
+                            "identity_alias_accepted": False,
+                            "secrets_included": False,
+                        }
+                    )
+                except Exception:
+                    pass
+            self.send_json(exc.status, {"error": exc.code, **exc.details})
+        except Exception:
+            self.send_json(500, {"error": "writer_route_internal_error"})
+
+
+def main() -> None:
+    load_binding()
+    server = ThreadingHTTPServer((HOST, PORT), Handler)
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()

active-runtime/app/aios_isolated_writer_api/__init__.py

@@ -0,0 +1,23 @@
+"""Task-bound Integration Refresh control-plane primitives."""
+
+from .integration_merge_writer import IntegrationMergeWriter, MergeError, MergeRequest
+from .integration_refresh import (
+    AppendOnlyAudit,
+    Binding,
+    GiteaBranchReadProjection,
+    GrantStore,
+    IntegrationRefreshProvisioner,
+    ProvisioningError,
+)
+
+__all__ = [
+    "AppendOnlyAudit",
+    "Binding",
+    "GiteaBranchReadProjection",
+    "GrantStore",
+    "IntegrationMergeWriter",
+    "IntegrationRefreshProvisioner",
+    "MergeError",
+    "MergeRequest",
+    "ProvisioningError",
+]

active-runtime/app/aios_isolated_writer_api/integration_merge_writer.py

@@ -0,0 +1,90 @@
+"""Constrained integration-only merge writer with rollback and reconciliation states."""
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+import re
+
+
+class MergeError(RuntimeError):
+    def __init__(self, code: str, status: int = 409):
+        super().__init__(code); self.code, self.status = code, status
+
+
+@dataclass(frozen=True)
+class MergeRequest:
+    integration_branch: str
+    source_branch: str
+    approved_source_oid: str
+
+
+class IntegrationMergeWriter:
+    def __init__(self, workspace: Path | str, remote: str, projection):
+        self.workspace, self.remote, self.projection = Path(workspace), remote, projection
+
+    def _git(self, *args: str, check=True) -> str:
+        p = subprocess.run(["git", *args], cwd=self.workspace, text=True, capture_output=True)
+        if check and p.returncode:
+            raise MergeError("git_operation_failed", 500)
+        return p.stdout.strip()
+
+    def merge(self, request: MergeRequest) -> dict:
+        def bad(value: str) -> bool:
+            return (
+                value == "main"
+                or value.startswith("main/")
+                or value == "governance"
+                or value.startswith("governance/")
+                or value == "approvals"
+                or value.startswith("refs/")
+                or value.startswith("tags/")
+            )
+        if (bad(request.integration_branch) or bad(request.source_branch) or
+                not request.integration_branch.startswith("integration/") or
+                re.fullmatch(r"[0-9a-fA-F]{40}", request.approved_source_oid) is None):
+            raise MergeError("integration_merge_scope_rejected", 403)
+        for branch in (request.integration_branch, request.source_branch):
+            if subprocess.run(["git", "check-ref-format", "--branch", branch], cwd=self.workspace,
+                              text=True, capture_output=True).returncode:
+                raise MergeError("integration_merge_scope_rejected", 403)
+        if self._git("rev-parse", "--abbrev-ref", "HEAD") != request.integration_branch:
+            raise MergeError("integration_branch_checkout_required")
+        before = self._git("rev-parse", "HEAD")
+        remote_before = self.projection(request.integration_branch)
+        if remote_before != before:
+            raise MergeError("integration_remote_advanced")
+        source = self._git("rev-parse", "--verify", f"refs/heads/{request.source_branch}^{{commit}}")
+        if source != request.approved_source_oid:
+            raise MergeError("integration_source_not_approved", 403)
+        attempt = subprocess.run(["git", "merge", "--no-ff", "--no-commit", request.source_branch], cwd=self.workspace, text=True, capture_output=True)
+        if attempt.returncode:
+            subprocess.run(["git", "merge", "--abort"], cwd=self.workspace, text=True, capture_output=True)
+            raise MergeError("integration_merge_conflict")
+        try:
+            self._git("commit", "-m", "merge(integration): approved source")
+            merged = self._git("rev-parse", "HEAD")
+            parents = self._git("show", "-s", "--format=%P", "HEAD").split()
+            if parents != [before, source]:
+                raise MergeError("integration_parent_invariant_failed", 500)
+            pushed = subprocess.run(["git", "push", "--porcelain", self.remote, f"HEAD:refs/heads/{request.integration_branch}"], cwd=self.workspace, text=True, capture_output=True)
+            if pushed.returncode:
+                try:
+                    observed = self.projection(request.integration_branch)
+                except Exception:
+                    return {"state": "RECONCILIATION_REQUIRED", "pre_merge_oid": before, "merge_oid": merged}
+                if observed == before:
+                    self._git("reset", "--hard", before)
+                    return {"state": "PUSH_FAILED_RESTORED", "pre_merge_oid": before}
+                return {"state": "RECONCILIATION_REQUIRED", "pre_merge_oid": before, "merge_oid": merged}
+            try:
+                observed = self.projection(request.integration_branch)
+            except Exception:
+                return {"state": "RECONCILIATION_REQUIRED", "pre_merge_oid": before, "merge_oid": merged}
+            if observed != merged:
+                return {"state": "RECONCILIATION_REQUIRED", "pre_merge_oid": before, "merge_oid": merged}
+            return {"state": "COMPLETED", "pre_merge_oid": before, "merge_oid": merged}
+        except MergeError:
+            if self._git("rev-parse", "HEAD") != before:
+                self._git("reset", "--hard", before)
+            raise

active-runtime/app/aios_isolated_writer_api/integration_refresh.py

@@ -0,0 +1,257 @@
+"""Fail-closed, task-bound integration refresh provisioning primitives.
+
+This module deliberately has no HTTP or credential-provider dependency.  The
+caller supplies a branch-read projection and an apply callback, making the
+authorization and recovery boundary independently testable.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import argparse
+import secrets
+import sqlite3
+import threading
+import re
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Callable
+
+
+class ProvisioningError(RuntimeError):
+    def __init__(self, code: str, status: int = 403):
+        super().__init__(code)
+        self.code, self.status = code, status
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _digest(secret: str) -> str:
+    return hashlib.sha256(secret.encode("utf-8")).hexdigest()
+
+
+def _canonical(value: dict) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def _is_oid(value: object) -> bool:
+    return isinstance(value, str) and re.fullmatch(r"[0-9a-fA-F]{40}", value) is not None
+
+
+@dataclass(frozen=True)
+class Binding:
+    task_id: str
+    repository: str
+    branch: str
+    actor_identity: str
+    audience: str
+    paths: tuple[str, ...]
+    quota: int
+
+
+class AppendOnlyAudit:
+    """Append only redacted JSONL audit, durable before a result is returned."""
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    def append(self, event: dict) -> None:
+        with self._lock:
+            safe = {key: value for key, value in event.items() if key not in {"token", "credential", "authorization"}}
+            previous = "0" * 64
+            if self.path.exists():
+                lines = self.path.read_text(encoding="utf-8").splitlines()
+                if lines:
+                    previous = json.loads(lines[-1])["event_hash"]
+            safe["previous_event_hash"] = previous
+            safe["event_hash"] = _canonical(safe)
+            encoded = (json.dumps(safe, sort_keys=True, separators=(",", ":")) + "\n").encode()
+            fd = os.open(self.path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600)
+            try:
+                os.write(fd, encoded)
+                os.fsync(fd)
+            finally:
+                os.close(fd)
+
+
+class GrantStore:
+    """Digest-only short grant registry; SQLite transactions serialize quota use."""
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self._db() as db:
+            db.execute("""CREATE TABLE IF NOT EXISTS grants(
+                token_digest TEXT PRIMARY KEY, binding_json TEXT NOT NULL,
+                expires_at TEXT NOT NULL, quota INTEGER NOT NULL, uses INTEGER NOT NULL,
+                revoked INTEGER NOT NULL DEFAULT 0)""")
+
+    @contextmanager
+    def _db(self):
+        db = sqlite3.connect(self.path, timeout=15, isolation_level=None)
+        db.row_factory = sqlite3.Row
+        try:
+            yield db
+        finally:
+            db.close()
+
+    def issue(self, binding: Binding, *, ttl_seconds: int = 300) -> str:
+        if ttl_seconds < 1 or ttl_seconds > 1800 or binding.quota < 1 or not binding.paths:
+            raise ValueError("invalid_short_grant")
+        token = secrets.token_urlsafe(32)
+        payload = json.dumps(binding.__dict__ | {"paths": list(binding.paths)}, sort_keys=True)
+        with self._db() as db:
+            db.execute("INSERT INTO grants VALUES(?,?,?,?,?,0)", (_digest(token), payload,
+                (_now() + timedelta(seconds=ttl_seconds)).isoformat(), binding.quota, 0))
+        return token
+
+    def rotate(self, old_token: str, binding: Binding, *, revocation_probe, ttl_seconds: int = 300) -> str:
+        self.revoke(old_token)
+        try: status = revocation_probe(old_token)
+        except TimeoutError as exc: raise ProvisioningError("revocation_probe_timeout", 503) from exc
+        if status != 401: raise ProvisioningError("revocation_probe_rejected", 503)
+        return self.issue(binding, ttl_seconds=ttl_seconds)
+
+    def revoke(self, token: str) -> None:
+        with self._db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            db.execute("UPDATE grants SET revoked=1 WHERE token_digest=?", (_digest(token),))
+            db.execute("COMMIT")
+
+    def consume(self, token: str, binding: Binding, path: str) -> None:
+        with self._db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT * FROM grants WHERE token_digest=?", (_digest(token),)).fetchone()
+            if not row or row["revoked"]:
+                db.execute("ROLLBACK"); raise ProvisioningError("credential_rejected", 401)
+            if _now() >= datetime.fromisoformat(row["expires_at"]):
+                db.execute("ROLLBACK"); raise ProvisioningError("credential_expired", 401)
+            stored = json.loads(row["binding_json"])
+            requested = binding.__dict__ | {"paths": list(binding.paths)}
+            if stored != requested or path not in stored["paths"]:
+                db.execute("ROLLBACK"); raise ProvisioningError("grant_binding_rejected")
+            if row["uses"] >= row["quota"]:
+                db.execute("ROLLBACK"); raise ProvisioningError("grant_quota_exhausted")
+            db.execute("UPDATE grants SET uses=uses+1 WHERE token_digest=?", (_digest(token),))
+            db.execute("COMMIT")
+
+
+class GiteaBranchReadProjection:
+    """Normalizes a trusted task-scoped Gitea reader; every ambiguity fails closed."""
+    def __init__(self, reader: Callable[[str, str], dict]):
+        self.reader = reader
+
+    def head(self, repository: str, branch: str) -> str:
+        try:
+            reply = self.reader(repository, branch)
+        except TimeoutError as exc:
+            raise ProvisioningError("gitea_projection_timeout", 503) from exc
+        except Exception as exc:
+            raise ProvisioningError("gitea_projection_unavailable", 503) from exc
+        if not isinstance(reply, dict) or reply.get("status") in {401, 403}:
+            raise ProvisioningError("gitea_projection_forbidden", 503)
+        oid = reply.get("oid")
+        if reply.get("status") != 200 or not _is_oid(oid):
+            raise ProvisioningError("gitea_projection_invalid", 503)
+        return oid.lower()
+
+
+class IntegrationRefreshProvisioner:
+    _locks: dict[tuple[str, str], threading.Lock] = {}
+    _locks_guard = threading.Lock()
+
+    def __init__(self, *, grants: GrantStore, projection: GiteaBranchReadProjection,
+                 ledger_path: Path | str, audit: AppendOnlyAudit, approved_binding: Binding):
+        self.grants, self.projection, self.audit, self.approved_binding = grants, projection, audit, approved_binding
+        self.ledger_path = Path(ledger_path); self.ledger_path.parent.mkdir(parents=True, exist_ok=True)
+        with self._ledger() as db:
+            db.execute("CREATE TABLE IF NOT EXISTS refreshes(operation_id TEXT PRIMARY KEY, digest TEXT NOT NULL, state TEXT NOT NULL, result_oid TEXT)")
+
+    @contextmanager
+    def _ledger(self):
+        db = sqlite3.connect(self.ledger_path)
+        try:
+            yield db
+            db.commit()
+        finally:
+            db.close()
+
+    def refresh(self, *, token: str, binding: Binding, path: str, operation_id: str,
+                target_head_oid: str, apply: Callable[[], str]) -> dict:
+        if not operation_id or not _is_oid(target_head_oid):
+            raise ProvisioningError("refresh_request_rejected")
+        if binding != self.approved_binding:
+            raise ProvisioningError("grant_binding_rejected")
+        request = {"binding": binding.__dict__ | {"paths": list(binding.paths)}, "path": path, "target": target_head_oid}
+        digest = _canonical(request)
+        with self._locks_guard:
+            lock = self._locks.setdefault((binding.repository, binding.branch), threading.Lock())
+        with lock, self._ledger() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT * FROM refreshes WHERE operation_id=?", (operation_id,)).fetchone()
+            new_operation = row is None
+            if row:
+                if row[1] != digest:
+                    db.execute("ROLLBACK"); raise ProvisioningError("operation_id_reuse_rejected", 409)
+                if row[2] == "COMPLETED":
+                    db.execute("COMMIT")
+                    return {"state": row[2], "operation_id": operation_id, "oid": row[3], "idempotent": True}
+                if row[2] not in {"PENDING", "RECOVERABLE"}:
+                    db.execute("COMMIT")
+                    return {"state": row[2], "operation_id": operation_id, "oid": row[3], "idempotent": True}
+            else:
+                db.execute("INSERT INTO refreshes VALUES(?,?,?,NULL)", (operation_id, digest, "PENDING"))
+            db.execute("COMMIT")
+            if new_operation:
+                self.grants.consume(token, binding, path)
+            current = self.projection.head(binding.repository, binding.branch)
+            if current != target_head_oid.lower():
+                self._state(operation_id, "STALE_TARGET")
+                raise ProvisioningError("stale_target_head", 409)
+            try:
+                oid = apply()
+                if not _is_oid(oid):
+                    raise ProvisioningError("apply_result_invalid", 500)
+                # Detect a competing remote advancement before claiming completion.
+                observed = self.projection.head(binding.repository, binding.branch)
+                if observed != oid.lower():
+                    self._state(operation_id, "RECONCILIATION_REQUIRED")
+                    raise ProvisioningError("remote_concurrent_advance", 409)
+                self._state(operation_id, "COMPLETED", oid)
+                self.audit.append({"operation_id": operation_id, "task_id": binding.task_id, "repository": binding.repository,
+                                   "branch": binding.branch, "actor_identity": binding.actor_identity, "audience": binding.audience, "path": path, "state": "COMPLETED"})
+                return {"state": "COMPLETED", "operation_id": operation_id, "oid": oid, "idempotent": False}
+            except ProvisioningError:
+                raise
+            except Exception as exc:
+                self._state(operation_id, "RECOVERABLE")
+                self.audit.append({"operation_id": operation_id, "task_id": binding.task_id, "repository": binding.repository,
+                                   "branch": binding.branch, "actor_identity": binding.actor_identity, "audience": binding.audience, "path": path, "state": "RECOVERABLE"})
+                raise ProvisioningError("refresh_apply_failed", 503) from exc
+
+    def _state(self, operation_id: str, state: str, oid: str | None = None) -> None:
+        with self._ledger() as db:
+            db.execute("UPDATE refreshes SET state=?, result_oid=COALESCE(?,result_oid) WHERE operation_id=?", (state, oid, operation_id))
+
+
+def _sandbox_preflight(state_dir: Path) -> int:
+    if os.environ.get("AIOS_GITOPS_WRITES_ENABLED") != "false":
+        raise ProvisioningError("global_gitops_writes_must_remain_disabled", 500)
+    state_dir.mkdir(parents=True, exist_ok=True)
+    audit = AppendOnlyAudit(state_dir / "preflight-audit.jsonl")
+    audit.append({"event": "SANDBOX_PREFLIGHT", "writes_enabled": False})
+    print(json.dumps({"status": "READY", "mode": "sandbox", "writes_enabled": False}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--sandbox-preflight", action="store_true", required=True)
+    parser.add_argument("--state-dir", type=Path, required=True)
+    options = parser.parse_args()
+    raise SystemExit(_sandbox_preflight(options.state_dir))

deploy/systemd/aios-governance-machine-status-producer.service

@@ -1,26 +0,0 @@
-[Unit]
-Description=AIOS dormant single-owner machine-status prerequisite
-RefuseManualStop=no
-
-[Service]
-Type=oneshot
-User=aiman
-Group=gitops
-Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
-Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
-Environment=AIOS_GITOPS_WRITES_ENABLED=false
-ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component machine-status-producer --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
-NoNewPrivileges=true
-PrivateTmp=true
-PrivateDevices=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectControlGroups=true
-RestrictSUIDSGID=true
-LockPersonality=true
-MemoryDenyWriteExecute=true
-RestrictAddressFamilies=AF_UNIX
-ReadOnlyPaths=/data/aios-gitops/repo.git
-ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance

deploy/systemd/aios-governance-owner-review-adapter.service

@@ -1,26 +0,0 @@
-[Unit]
-Description=AIOS dormant single-owner review adapter prerequisite
-RefuseManualStop=no
-
-[Service]
-Type=oneshot
-User=aiman
-Group=gitops
-Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
-Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
-Environment=AIOS_GITOPS_WRITES_ENABLED=false
-ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component owner-review-adapter --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
-NoNewPrivileges=true
-PrivateTmp=true
-PrivateDevices=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectControlGroups=true
-RestrictSUIDSGID=true
-LockPersonality=true
-MemoryDenyWriteExecute=true
-RestrictAddressFamilies=AF_UNIX
-ReadOnlyPaths=/data/aios-gitops/repo.git
-ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance

deploy/systemd/aios-governance-single-owner-activation.service

@@ -1,20 +0,0 @@
-[Unit]
-Description=AIOS single-owner governance activation verifier
-ConditionPathExists=/etc/aios-gitops/single-owner-activation-bindings.json
-
-[Service]
-Type=oneshot
-User=root
-Group=gitops
-Environment=AIOS_GITOPS_WRITES_ENABLED=false
-ExecStart=/usr/bin/aios-single-owner-governance --verify --bindings /etc/aios-gitops/single-owner-activation-bindings.json --state-root /var/lib/aios-gitops/governance-activation --lock-path /run/aios-gitops/governance-activation.lock
-NoNewPrivileges=true
-PrivateTmp=true
-ProtectHome=true
-ProtectSystem=strict
-ReadWritePaths=/var/lib/aios-gitops/governance-activation /run/aios-gitops
-RestrictSUIDSGID=true
-UMask=0027
-
-# Deliberately not installable. Deployment remains disabled/inactive;
-# activation is an explicit approved CLI transaction, never an always-on unit.

deploy/systemd/aios-integration-refresh-provisioner.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=AIOS Integration Refresh Provisioner (sandbox control plane)
+After=network-online.target
+
+[Service]
+Type=oneshot
+RemainAfterExit=yes
+User=aios
+Group=aios
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_INTEGRATION_REFRESH_MODE=sandbox
+Environment=AIOS_CREDENTIAL_PROVIDER=disabled
+NoNewPrivileges=yes
+PrivateTmp=yes
+ProtectSystem=strict
+ProtectHome=yes
+ReadWritePaths=/var/lib/aios/integration-refresh
+RestrictAddressFamilies=AF_UNIX
+IPAddressDeny=any
+ExecStart=/usr/bin/python3 /opt/aios/active-runtime/app/aios_isolated_writer_api/integration_refresh.py --sandbox-preflight --state-dir /var/lib/aios/integration-refresh
+
+[Install]
+WantedBy=multi-user.target

deploy/tmpfiles.d/aios-single-owner-governance.conf

@@ -1,4 +0,0 @@
-d /var/lib/aios-gitops/governance-activation 0750 root gitops -
-d /var/lib/aios-gitops/governance-activation/wal 0750 root gitops -
-d /var/lib/aios-gitops/governance-activation/receipts 0750 root gitops -
-d /run/aios-gitops 0750 root gitops -

docs/autonomous-release/INTEGRATION_REFRESH_SOURCE_MAP_20260728.json

@@ -0,0 +1,35 @@
+{
+  "execution_renewal_manifest_blob_oid": "417f78abb3d795f7b0c04a0894069bd166bb15df",
+  "execution_renewal_manifest_sha256": "069091eff9c7c46bceccde365da6d07a8538e76bee9e288a57211f3114b94ab3",
+  "execution_renewal_merge_oid": "3a22c3c7d86930efd4bd93ac05d80ad322050cb9",
+  "observed_at": "2026-07-28T09:09:27.841968Z",
+  "runtime_source_evidence_sha256": "23f03d9c6abe6f11e1acda5210b8708696e3b77caa0083b6bfaef30a44b2a35f",
+  "runtime_source_observed_at": "2026-07-28T05:53:01.253909Z",
+  "schema": "aios.integration-refresh-source-map/v1",
+  "source_mapping_approval_merge_oid": "9514062cf33e45941abe4a8c98377e0713b0b3f6",
+  "source_mapping_manifest_blob_oid": "00c5e90e529cb98c7750a5761b0a0d6ccf507174",
+  "source_mapping_manifest_sha256": "17af8ebc43673c0e46bb9ab1bf44497e8c147f1b4a9d7a8a404edd5c0abd6f21",
+  "sources": [
+    {
+      "byte_for_byte_equal": true,
+      "canonical_blob_oid": "9c68be36e0b6a18260e3099e3d7b0dcab6d814aa",
+      "canonical_commit_oid": "85426f4f570e4d40b212318d47e52e4aed5cb06a",
+      "canonical_git_path": "active-runtime/app/aios_autonomous_release_authorization/controller.py",
+      "canonical_tree_oid": "cdbd9bb845b18d4b5224d91aace2656a51eb7c14",
+      "runtime_blob_oid": "9c68be36e0b6a18260e3099e3d7b0dcab6d814aa",
+      "runtime_path": "/usr/libexec/aios-autonomous-release-authorization/controller.py",
+      "runtime_sha256": "39b532c3058c4ae49ed691d7c66c3b7745a25111fa61039f388c8335fd262c50"
+    },
+    {
+      "byte_for_byte_equal": true,
+      "canonical_blob_oid": "ee0595ca61a0598f8cb3dc10865a255e21679b2a",
+      "canonical_commit_oid": "85426f4f570e4d40b212318d47e52e4aed5cb06a",
+      "canonical_git_path": "active-runtime/app/aios_chatgpt_action_task_router/action_task_router.py",
+      "canonical_tree_oid": "cdbd9bb845b18d4b5224d91aace2656a51eb7c14",
+      "runtime_blob_oid": "ee0595ca61a0598f8cb3dc10865a255e21679b2a",
+      "runtime_path": "/usr/libexec/aios-chatgpt-action-task-router-continuous-20260726/action_task_router.py",
+      "runtime_sha256": "738fb7035a6b792115d6467b4d1e4ffd47a8c5d5646f7032ebc68ea587cbd554"
+    }
+  ],
+  "task_id": "AIOS-INTEGRATION-REFRESH-PROVISIONER-IMPLEMENTATION-20260726"
+}

evidence/AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729/live-activation-evidence-closure.json

@@ -1,50 +0,0 @@
-{
-  "artifact_sha256": "01e0226f0302bb8e2f8d81299393075bf0332c6e22e00877469fb297c51f3802",
-  "branch": "patch/single-owner-governance-live-activation-20260729",
-  "controller_visible_governance_oid": "8b7f9d9028f6ff9beb225bf85f53dbeaac6830e7",
-  "created_at": "2026-07-29T03:28:15Z",
-  "implementation_commit_oid": "e1e438564ce8b987a36df0947e6eaa324dfede74",
-  "implementation_tree_oid": "dd93dc1df10cb8b8fe59b09d3214a9ae6fee671e",
-  "independent_validation": {
-    "activation_py_sha256": "86cde2079b4f3493957ba346e71b02b387693e63a3f38be418710ff7c348d19a",
-    "approval_backend_dual_database": "DENIED",
-    "expiry_replay": "DENIED",
-    "operation_replay": "DENIED",
-    "result": "PASS",
-    "target_replay": "DENIED",
-    "task_replay": "DENIED"
-  },
-  "mirror_closure": {
-    "pr30_merge_oid": "f6bef90633f428454e85721f353cd82cdf04caf0",
-    "pr31_merge_oid": "8b7f9d9028f6ff9beb225bf85f53dbeaac6830e7",
-    "receipt_sha256": "b3edc4eb0701efcb448b981029ab9a67a0637f8349bf4a047f430f647e4a8fd0",
-    "result": "PASS"
-  },
-  "production_isolation": {
-    "aios_gitops_writes_enabled": false,
-    "live_activation_service_active": false,
-    "live_activation_service_enabled": false,
-    "main_modified": false,
-    "production_modified": false,
-    "protected_refs_unchanged": true
-  },
-  "rollback_artifact_sha256": "ed9307cc2f07767b75866fcb370fa3f98b76c5147e4b7fc49ecf04cf21cf7c31",
-  "signed_merge_audit": {
-    "merge_signature_policy": "NOT_REQUIRED",
-    "pr31_signature_policy_result": "UNSIGNED_BUT_POLICY_VALID",
-    "signed_merge_requirement_effective": false
-  },
-  "task_id": "AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729",
-  "test_evidence_sha256": "172a7cf812c6b901524138fb72851a9b7d2e895dedc98893407d0bab9b52c6c1",
-  "tests": {
-    "collection_errors": 0,
-    "compileall": "PASS",
-    "copy_of_live_gitea_acceptance": "PASS",
-    "secret_scan": "PASS",
-    "subtests_passed": 17,
-    "tests_failed": 0,
-    "tests_passed": 75,
-    "transaction_tests": "PASS",
-    "rollback_tests": "PASS"
-  }
-}

services/aios-single-owner-governance/README.md

@@ -1,24 +0,0 @@
-# AIOS Single-Owner Governance Prerequisites
-
-This package is a dormant candidate produced under the PR #29 prerequisite
-authorization. It validates exact-head Owner approvals and machine gates,
-models protection-policy compare-and-swap with compensation, builds evidence,
-and exercises every mutation path only against disposable fixtures.
-
-The prerequisite shadow services remain static and refuse live mutation. This
-branch adds a separate, one-shot activation executor with `--plan`,
-`--activate`, `--verify`, `--rollback`, and `--status` commands. It is not an
-always-on writer and never enables global GitOps writes.
-
-The executor is fail-closed: every consequential command is bound to an exact
-approval merge, governance OID, evidence closure, Owner review, component
-digests, full branch-protection snapshot, operation ID, and plan ID. It writes
-a fsynced WAL before the single-row SQLite CAS, verifies the full applied row,
-and only rolls back with a target-to-previous CAS. Audit records are
-hash-chained and credentials are redacted.
-
-Installation must remain disabled and inactive until a merged
-`SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION`
-approval authorizes one explicit execution. The approved runtime paths are
-`/var/lib/aios-gitops/governance-activation/` and
-`/run/aios-gitops/governance-activation.lock`.

services/aios-single-owner-governance/activation-bindings.schema.json

@@ -1,80 +0,0 @@
-{
-  "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "additionalProperties": false,
-  "properties": {
-    "activation_approval_merge_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "activation_approval_pr_number": {"minimum": 1, "type": "integer"},
-    "activation_manifest_blob_oid": {"pattern": "^[0-9a-f]{40,64}$", "type": "string"},
-    "activation_manifest_path": {"pattern": "^approvals/.+\\.json$", "type": "string"},
-    "activation_manifest_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "approval_db_path": {"minLength": 1, "type": "string"},
-    "approval_request_head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "backend": {"enum": ["file", "sqlite"]},
-    "backend_path": {"minLength": 1, "type": "string"},
-    "branch": {"const": "governance/approvals"},
-    "evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "evidence_git_dir": {"minLength": 1, "type": "string"},
-    "executor": {"not": {"const": "owner"}, "type": "string"},
-    "expected_branch_protection_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "expected_gitea_schema_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "expected_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "expected_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "expected_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "expires_at": {"format": "date-time", "type": "string"},
-    "forbidden_operations": {"additionalProperties": {"const": false}, "type": "object"},
-    "governance_git_dir": {"minLength": 1, "type": "string"},
-    "governance_ref": {"const": "refs/heads/governance/approvals"},
-    "machine_checks": {
-      "additionalProperties": false,
-      "properties": {
-        "aios/branch-protection": {"$ref": "#/$defs/machineCheck"},
-        "aios/evidence-closure": {"$ref": "#/$defs/machineCheck"},
-        "aios/secret-scan": {"$ref": "#/$defs/machineCheck"},
-        "aios/state-machine": {"$ref": "#/$defs/machineCheck"},
-        "aios/validator": {"$ref": "#/$defs/machineCheck"}
-      },
-      "required": ["aios/branch-protection", "aios/evidence-closure", "aios/secret-scan", "aios/state-machine", "aios/validator"],
-      "type": "object"
-    },
-    "observed_evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "observed_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-    "observed_owner_review_oid": {"type": ["integer", "string"]},
-    "observed_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "observed_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
-    "operation_id": {"minLength": 1, "type": "string"},
-    "owner_review_oid": {"type": ["integer", "string"]},
-    "repo_id": {"minimum": 1, "type": "integer"},
-    "requester": {"not": {"const": "owner"}, "type": "string"},
-    "state_machine_path": {"minLength": 1, "type": "string"},
-    "target_branch_protection": {"type": "object"},
-    "task_id": {"minLength": 1, "type": "string"},
-    "validator_path": {"minLength": 1, "type": "string"},
-    "wip": {"const": false}
-  },
-  "$defs": {
-    "machineCheck": {
-      "additionalProperties": false,
-      "properties": {
-        "head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
-        "state": {"const": "success"}
-      },
-      "required": ["state", "head_oid"],
-      "type": "object"
-    }
-  },
-  "required": [
-    "task_id", "operation_id", "expected_governance_oid", "observed_governance_oid",
-    "expected_validator_sha256", "observed_validator_sha256", "validator_path",
-    "expected_state_machine_sha256", "observed_state_machine_sha256", "state_machine_path",
-    "expected_branch_protection_sha256", "expected_gitea_schema_sha256",
-    "activation_approval_merge_oid", "activation_approval_pr_number", "approval_request_head_oid",
-    "activation_manifest_path", "activation_manifest_blob_oid", "activation_manifest_sha256",
-    "evidence_closure_oid", "observed_evidence_closure_oid", "evidence_git_dir",
-    "owner_review_oid", "observed_owner_review_oid", "approval_db_path", "expires_at",
-    "governance_git_dir", "governance_ref", "machine_checks", "forbidden_operations",
-    "target_branch_protection", "backend", "backend_path", "repo_id", "branch",
-    "requester", "executor", "wip"
-  ],
-  "title": "AIOS Single Owner Live Activation Bindings",
-  "type": "object"
-}

services/aios-single-owner-governance/aios_single_owner_governance/__init__.py

@@ -1,20 +0,0 @@
-"""Dormant single-owner governance activation prerequisites."""
-
-from .models import (
-    ApprovalContext,
-    GovernanceState,
-    MachineGate,
-    ProtectionSnapshot,
-    Review,
-)
-from .validator import SingleOwnerApprovalValidator, ValidationDenied
-
-__all__ = [
-    "ApprovalContext",
-    "GovernanceState",
-    "MachineGate",
-    "ProtectionSnapshot",
-    "Review",
-    "SingleOwnerApprovalValidator",
-    "ValidationDenied",
-]

services/aios-single-owner-governance/aios_single_owner_governance/activation.py

@@ -1,686 +0,0 @@
-from __future__ import annotations
-
-import argparse
-from contextlib import contextmanager
-from datetime import datetime, timezone
-import hashlib
-import json
-import os
-from pathlib import Path
-import re
-import sqlite3
-import subprocess
-from typing import Any, Iterator, Mapping, Protocol
-
-try:
-    import fcntl
-except ImportError:  # pragma: no cover
-    fcntl = None
-
-from .gitea_protected_branch import (
-    CASConflict,
-    FileProtectedBranchAdapter,
-    SQLiteProtectedBranchAdapter,
-    calculate_snapshot_sha256,
-)
-from .journal import JournalStore, atomic_write_json, read_json, redact, sha256_json
-
-
-DEFAULT_STATE_ROOT = Path("/var/lib/aios-gitops/governance-activation")
-DEFAULT_LOCK_PATH = Path("/run/aios-gitops/governance-activation.lock")
-LIVE_GITEA_DB = Path("/data/aios-gitops/runtime/gitea-bin/data/gitea.db")
-LIVE_AUTHORITATIVE_GIT = Path("/data/aios-gitops/repo.git")
-HEX40 = re.compile(r"^[0-9a-f]{40}$")
-HEX64 = re.compile(r"^[0-9a-f]{64}$")
-REQUIRED_BINDINGS = {
-    "task_id",
-    "operation_id",
-    "expected_governance_oid",
-    "expected_validator_sha256",
-    "expected_state_machine_sha256",
-    "expected_branch_protection_sha256",
-    "activation_approval_merge_oid",
-    "evidence_closure_oid",
-    "owner_review_oid",
-    "observed_governance_oid",
-    "observed_validator_sha256",
-    "observed_state_machine_sha256",
-    "observed_evidence_closure_oid",
-    "observed_owner_review_oid",
-    "approval_request_head_oid",
-    "activation_approval_pr_number",
-    "expires_at",
-    "validator_path",
-    "state_machine_path",
-    "governance_git_dir",
-    "governance_ref",
-    "evidence_git_dir",
-    "approval_db_path",
-    "expected_gitea_schema_sha256",
-    "backend",
-    "backend_path",
-    "repo_id",
-    "branch",
-    "activation_manifest_path",
-    "activation_manifest_blob_oid",
-    "activation_manifest_sha256",
-}
-REQUIRED_MACHINE_CONTEXTS = {
-    "aios/validator",
-    "aios/state-machine",
-    "aios/branch-protection",
-    "aios/secret-scan",
-    "aios/evidence-closure",
-}
-
-
-class InjectedCrash(BaseException):
-    pass
-
-
-class Adapter(Protocol):
-    def read_current(self) -> dict[str, Any]: ...
-    def calculate_snapshot_sha256(self, snapshot: Mapping[str, Any]) -> str: ...
-    def apply_atomic(self, expected_sha256: str, target: Mapping[str, Any]) -> dict[str, Any]: ...
-    def verify_applied(self, expected_sha256: str) -> bool: ...
-    def rollback_atomic(self, expected_target_sha256: str, previous: Mapping[str, Any]) -> dict[str, Any]: ...
-    def verify_rollback(self, expected_sha256: str) -> bool: ...
-
-
-@contextmanager
-def exclusive_lock(path: Path) -> Iterator[None]:
-    path.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
-    flags = os.O_RDWR | os.O_CREAT
-    if hasattr(os, "O_NOFOLLOW"):
-        flags |= os.O_NOFOLLOW
-    fd = os.open(path, flags, 0o640)
-    try:
-        os.fchmod(fd, 0o640)
-        if fcntl is not None:
-            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        yield
-    finally:
-        if fcntl is not None:
-            fcntl.flock(fd, fcntl.LOCK_UN)
-        os.close(fd)
-
-
-def _require_hex(name: str, value: object, lengths: tuple[int, ...]) -> None:
-    if not isinstance(value, str) or len(value) not in lengths or not re.fullmatch(r"[0-9a-f]+", value):
-        raise RuntimeError(f"{name.upper()}_INVALID")
-
-
-def _sha256_file(path: Path) -> str:
-    if not path.is_absolute() or not path.is_file() or path.is_symlink():
-        raise RuntimeError("VERIFIED_COMPONENT_PATH_INVALID")
-    digest = hashlib.sha256()
-    with path.open("rb") as handle:
-        for block in iter(lambda: handle.read(1024 * 1024), b""):
-            digest.update(block)
-    return digest.hexdigest()
-
-
-def _git(git_dir: Path, *args: str) -> str:
-    if not git_dir.is_absolute() or not git_dir.exists() or git_dir.is_symlink():
-        raise RuntimeError("GIT_OBJECT_STORE_INVALID")
-    result = subprocess.run(
-        ["git", f"--git-dir={git_dir}", *args],
-        text=True,
-        capture_output=True,
-        check=False,
-    )
-    if result.returncode != 0:
-        raise RuntimeError("GIT_EVIDENCE_VERIFICATION_FAILED")
-    return result.stdout.strip()
-
-
-def _parse_utc(value: object) -> datetime:
-    if not isinstance(value, str):
-        raise RuntimeError("EXPIRY_INVALID")
-    try:
-        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
-    except ValueError as exc:
-        raise RuntimeError("EXPIRY_INVALID") from exc
-    if parsed.tzinfo is None:
-        raise RuntimeError("EXPIRY_INVALID")
-    return parsed.astimezone(timezone.utc)
-
-
-def _approved_execution_binding(bindings: Mapping[str, Any]) -> dict[str, Any]:
-    return {
-        "task_id": bindings["task_id"],
-        "operation_id": bindings["operation_id"],
-        "expected_validator_sha256": bindings["expected_validator_sha256"],
-        "expected_state_machine_sha256": bindings["expected_state_machine_sha256"],
-        "expected_branch_protection_sha256": bindings["expected_branch_protection_sha256"],
-        "expected_gitea_schema_sha256": bindings["expected_gitea_schema_sha256"],
-        "target_branch_protection_sha256": calculate_snapshot_sha256(
-            bindings["target_branch_protection"]
-        ),
-        "evidence_closure_oid": bindings["evidence_closure_oid"],
-        "backend": bindings["backend"],
-        "repo_id": bindings["repo_id"],
-        "branch": bindings["branch"],
-        "requester": bindings["requester"],
-        "executor": bindings["executor"],
-        "forbidden_operations": bindings["forbidden_operations"],
-    }
-
-
-def _same_path(left: Path, right: Path) -> bool:
-    try:
-        return left.resolve(strict=True) == right.resolve(strict=True) and os.path.samefile(left, right)
-    except (FileNotFoundError, OSError):
-        return False
-
-
-def _verify_authority_paths(bindings: Mapping[str, Any]) -> None:
-    if bindings["backend"] != "sqlite":
-        return
-    approval_db = Path(str(bindings["approval_db_path"]))
-    backend_db = Path(str(bindings["backend_path"]))
-    if not _same_path(approval_db, backend_db):
-        raise RuntimeError("APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH")
-    live = _same_path(backend_db, LIVE_GITEA_DB)
-    if live:
-        if not _same_path(Path(str(bindings["governance_git_dir"])), LIVE_AUTHORITATIVE_GIT):
-            raise RuntimeError("AUTHORITATIVE_GOVERNANCE_GIT_REQUIRED")
-        if not _same_path(Path(str(bindings["evidence_git_dir"])), LIVE_AUTHORITATIVE_GIT):
-            raise RuntimeError("AUTHORITATIVE_EVIDENCE_GIT_REQUIRED")
-        return
-    acceptance_root = os.environ.get("AIOS_GOVERNANCE_ACTIVATION_ACCEPTANCE_ROOT")
-    if not acceptance_root:
-        raise RuntimeError("NON_LIVE_SQLITE_BACKEND_DENIED")
-    root = Path(acceptance_root).resolve(strict=True)
-    if not str(root).startswith("/run/aios-live-activation-acceptance-"):
-        raise RuntimeError("ACCEPTANCE_ROOT_INVALID")
-    for path in (approval_db, backend_db, Path(str(bindings["governance_git_dir"])), Path(str(bindings["evidence_git_dir"]))):
-        try:
-            path.resolve(strict=True).relative_to(root)
-        except (FileNotFoundError, ValueError) as exc:
-            raise RuntimeError("ACCEPTANCE_AUTHORITY_ESCAPE_DENIED") from exc
-
-
-def _load_approved_manifest(bindings: Mapping[str, Any], git_dir: Path) -> dict[str, Any]:
-    path = str(bindings["activation_manifest_path"])
-    if not path.startswith("approvals/") or not path.endswith(".json") or ".." in Path(path).parts:
-        raise RuntimeError("ACTIVATION_MANIFEST_PATH_INVALID")
-    merge_oid = str(bindings["activation_approval_merge_oid"])
-    blob_oid = _git(git_dir, "rev-parse", f"{merge_oid}:{path}")
-    if blob_oid != bindings["activation_manifest_blob_oid"]:
-        raise RuntimeError("ACTIVATION_MANIFEST_BLOB_MISMATCH")
-    raw_result = subprocess.run(
-        ["git", f"--git-dir={git_dir}", "show", f"{merge_oid}:{path}"],
-        capture_output=True,
-        check=False,
-    )
-    if raw_result.returncode != 0:
-        raise RuntimeError("ACTIVATION_MANIFEST_READ_FAILED")
-    if hashlib.sha256(raw_result.stdout).hexdigest() != bindings["activation_manifest_sha256"]:
-        raise RuntimeError("ACTIVATION_MANIFEST_SHA256_MISMATCH")
-    try:
-        manifest = json.loads(raw_result.stdout)
-    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
-        raise RuntimeError("ACTIVATION_MANIFEST_INVALID") from exc
-    if not isinstance(manifest, dict):
-        raise RuntimeError("ACTIVATION_MANIFEST_INVALID")
-    return manifest
-
-
-def _verify_external_evidence(bindings: Mapping[str, Any]) -> None:
-    _verify_authority_paths(bindings)
-    validator_sha = _sha256_file(Path(str(bindings["validator_path"])))
-    state_machine_sha = _sha256_file(Path(str(bindings["state_machine_path"])))
-    if validator_sha != bindings["expected_validator_sha256"] or validator_sha != bindings["observed_validator_sha256"]:
-        raise RuntimeError("VALIDATOR_DIGEST_MISMATCH")
-    if state_machine_sha != bindings["expected_state_machine_sha256"] or state_machine_sha != bindings["observed_state_machine_sha256"]:
-        raise RuntimeError("STATE_MACHINE_DIGEST_MISMATCH")
-
-    governance_git_dir = Path(str(bindings["governance_git_dir"]))
-    governance_oid = _git(governance_git_dir, "rev-parse", "--verify", str(bindings["governance_ref"]))
-    if governance_oid != bindings["expected_governance_oid"] or governance_oid != bindings["observed_governance_oid"]:
-        raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
-    merge_oid = str(bindings["activation_approval_merge_oid"])
-    _git(governance_git_dir, "cat-file", "-e", f"{merge_oid}^{{commit}}")
-    if subprocess.run(
-        ["git", f"--git-dir={governance_git_dir}", "merge-base", "--is-ancestor", merge_oid, governance_oid],
-        capture_output=True,
-        check=False,
-    ).returncode != 0:
-        raise RuntimeError("ACTIVATION_APPROVAL_NOT_IN_GOVERNANCE_REF")
-    manifest = _load_approved_manifest(bindings, governance_git_dir)
-    if manifest.get("approval_type") != "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION":
-        raise RuntimeError("ACTIVATION_APPROVAL_TYPE_MISMATCH")
-    if manifest.get("execution_binding") != _approved_execution_binding(bindings):
-        raise RuntimeError("APPROVED_EXECUTION_BINDING_MISMATCH")
-    if manifest.get("execution_binding_sha256") != sha256_json(manifest["execution_binding"]):
-        raise RuntimeError("APPROVED_EXECUTION_BINDING_SHA256_MISMATCH")
-    if manifest.get("expires_at") != bindings["expires_at"]:
-        raise RuntimeError("APPROVAL_EXPIRY_BINDING_MISMATCH")
-
-    evidence_git_dir = Path(str(bindings["evidence_git_dir"]))
-    evidence_oid = str(bindings["evidence_closure_oid"])
-    if evidence_oid != bindings["observed_evidence_closure_oid"]:
-        raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
-    _git(evidence_git_dir, "cat-file", "-e", f"{evidence_oid}^{{object}}")
-
-    db_path = Path(str(bindings["approval_db_path"]))
-    if not db_path.is_absolute() or not db_path.is_file() or db_path.is_symlink():
-        raise RuntimeError("APPROVAL_DATABASE_INVALID")
-    expiry = _parse_utc(bindings["expires_at"])
-    if datetime.now(timezone.utc) >= expiry:
-        raise RuntimeError("APPROVAL_EXPIRED")
-    with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) as connection:
-        connection.row_factory = sqlite3.Row
-        pr = connection.execute(
-            "SELECT pr.issue_id,pr.has_merged,pr.merged_commit_id,i.is_closed "
-            "FROM pull_request pr JOIN issue i ON i.id=pr.issue_id "
-            "WHERE i.repo_id=? AND pr.\"index\"=? AND pr.base_branch=?",
-            (int(bindings["repo_id"]), int(bindings["activation_approval_pr_number"]), str(bindings["branch"])),
-        ).fetchone()
-        if pr is None or int(pr["has_merged"]) != 1 or int(pr["is_closed"]) != 1:
-            raise RuntimeError("ACTIVATION_APPROVAL_PR_NOT_MERGED")
-        if pr["merged_commit_id"] != bindings["activation_approval_merge_oid"]:
-            raise RuntimeError("ACTIVATION_APPROVAL_MERGE_MISMATCH")
-        review = connection.execute(
-            "SELECT r.id,r.type,r.reviewer_id,r.commit_id,r.stale,r.dismissed,r.created_unix,u.lower_name "
-            "FROM review r JOIN user u ON u.id=r.reviewer_id WHERE r.id=? AND r.issue_id=?",
-            (int(bindings["owner_review_oid"]), int(pr["issue_id"])),
-        ).fetchone()
-        if review is None or int(review["id"]) != int(bindings["observed_owner_review_oid"]):
-            raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
-        if (
-            int(review["type"]) != 1
-            or int(review["reviewer_id"]) != 2
-            or review["lower_name"] != "owner"
-            or review["commit_id"] != bindings["approval_request_head_oid"]
-            or int(review["stale"]) != 0
-            or int(review["dismissed"]) != 0
-            or datetime.fromtimestamp(int(review["created_unix"]), timezone.utc) >= expiry
-        ):
-            raise RuntimeError("OWNER_APPROVAL_INVALID")
-        statuses = connection.execute(
-            "SELECT id,context,state,sha FROM commit_status "
-            "WHERE repo_id=? AND sha=? ORDER BY id DESC",
-            (int(bindings["repo_id"]), str(bindings["approval_request_head_oid"])),
-        ).fetchall()
-        observed: dict[str, sqlite3.Row] = {}
-        for row in statuses:
-            if row["context"] in REQUIRED_MACHINE_CONTEXTS:
-                observed.setdefault(row["context"], row)
-        if set(observed) != REQUIRED_MACHINE_CONTEXTS:
-            raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
-        if any(row["state"] != "success" or row["sha"] != bindings["approval_request_head_oid"] for row in observed.values()):
-            raise RuntimeError("MACHINE_GATE_FAILED")
-
-
-def validate_bindings(bindings: Mapping[str, Any]) -> None:
-    missing = sorted(REQUIRED_BINDINGS - set(bindings))
-    if missing:
-        raise RuntimeError("MISSING_BINDINGS:" + ",".join(missing))
-    for name in (
-        "expected_governance_oid",
-        "activation_approval_merge_oid",
-        "evidence_closure_oid",
-    ):
-        _require_hex(name, bindings[name], (40,))
-    for name in (
-        "expected_validator_sha256",
-        "expected_state_machine_sha256",
-        "expected_branch_protection_sha256",
-        "observed_validator_sha256",
-        "observed_state_machine_sha256",
-        "expected_gitea_schema_sha256",
-        "activation_manifest_sha256",
-    ):
-        _require_hex(name, bindings[name], (64,))
-    _require_hex("activation_manifest_blob_oid", bindings["activation_manifest_blob_oid"], (40, 64))
-    if not isinstance(bindings["owner_review_oid"], (int, str)) or str(bindings["owner_review_oid"]).strip() == "":
-        raise RuntimeError("OWNER_REVIEW_OID_INVALID")
-    if bindings["governance_ref"] != "refs/heads/governance/approvals":
-        raise RuntimeError("GOVERNANCE_REF_INVALID")
-    if bindings["branch"] != "governance/approvals":
-        raise RuntimeError("BRANCH_BINDING_DENIED")
-    if bindings["backend"] not in {"file", "sqlite"}:
-        raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
-    if bindings["observed_governance_oid"] != bindings["expected_governance_oid"]:
-        raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
-    if bindings["observed_evidence_closure_oid"] != bindings["evidence_closure_oid"]:
-        raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
-    if str(bindings["observed_owner_review_oid"]) != str(bindings["owner_review_oid"]):
-        raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
-    if bindings.get("requester") == "owner":
-        raise RuntimeError("REQUESTER_SELF_APPROVAL_DENIED")
-    if bindings.get("executor") == "owner":
-        raise RuntimeError("EXECUTOR_SELF_APPROVAL_DENIED")
-    if bindings.get("wip"):
-        raise RuntimeError("WIP_BLOCKED")
-    checks = bindings.get("machine_checks")
-    if not isinstance(checks, dict) or set(checks) != REQUIRED_MACHINE_CONTEXTS:
-        raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
-    for context, check in checks.items():
-        if not isinstance(check, dict) or check.get("state") != "success" or check.get("head_oid") != bindings["approval_request_head_oid"]:
-            raise RuntimeError(f"MACHINE_GATE_FAILED:{context}")
-    forbidden = bindings.get("forbidden_operations", {})
-    if any(bool(value) for value in forbidden.values()):
-        raise RuntimeError("FORBIDDEN_OPERATION_REQUESTED")
-    if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
-        raise RuntimeError("GLOBAL_GITOPS_WRITES_MUST_REMAIN_FALSE")
-    if not isinstance(bindings.get("target_branch_protection"), dict):
-        raise RuntimeError("TARGET_BRANCH_PROTECTION_REQUIRED")
-    _verify_external_evidence(bindings)
-
-
-def binding_payload_hash(bindings: Mapping[str, Any]) -> str:
-    stable = dict(bindings)
-    stable.pop("crash_at", None)
-    return sha256_json(redact(stable))
-
-
-def plan_id(bindings: Mapping[str, Any], previous: Mapping[str, Any], target: Mapping[str, Any]) -> str:
-    stable = dict(bindings)
-    stable.pop("crash_at", None)
-    return sha256_json(
-        {
-            "bindings": redact(stable),
-            "previous": dict(previous),
-            "target": dict(target),
-        }
-    )
-
-
-class ActivationEngine:
-    def __init__(self, adapter: Adapter, state_root: Path, lock_path: Path) -> None:
-        self.adapter = adapter
-        self.journal = JournalStore(state_root)
-        self.lock_path = lock_path
-
-    def plan(self, bindings: Mapping[str, Any]) -> dict[str, Any]:
-        validate_bindings(bindings)
-        previous = self.adapter.read_current()
-        previous_sha = calculate_snapshot_sha256(previous)
-        if previous_sha != bindings["expected_branch_protection_sha256"]:
-            raise CASConflict("EXPECTED_BRANCH_PROTECTION_SHA256_MISMATCH")
-        target = dict(bindings["target_branch_protection"])
-        # Validate without writing by using the public target validator through a
-        # disposable comparison: adapters enforce the same checks again in CAS.
-        from .gitea_protected_branch import validate_target
-
-        validate_target(previous, target)
-        result = {
-            "status": "PLAN_READY",
-            "task_id": bindings["task_id"],
-            "operation_id": bindings["operation_id"],
-            "plan_id": plan_id(bindings, previous, target),
-            "payload_hash": binding_payload_hash(bindings),
-            "previous_sha256": previous_sha,
-            "target_sha256": calculate_snapshot_sha256(target),
-        }
-        return result
-
-    def _recover(self, record: dict[str, Any]) -> dict[str, Any]:
-        current = self.adapter.read_current()
-        current_sha = calculate_snapshot_sha256(current)
-        previous_sha = record["previous_sha256"]
-        target_sha = record["target_sha256"]
-        if record["phase"] == "PREPARED":
-            if current_sha == previous_sha:
-                return {**record, "recovery": "RESUME_BEFORE_APPLY"}
-            if current_sha == target_sha:
-                record = {**record, "phase": "APPLIED", "recovery": "RESUME_AFTER_APPLY"}
-                self.journal.write(record["operation_id"], record)
-                return record
-            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
-        if record["phase"] in {"APPLIED", "COMMITTED"} and current_sha not in {target_sha, previous_sha}:
-            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
-        return record
-
-    def _ensure_closure(self, record: dict[str, Any], event: str) -> None:
-        expected_sha = (
-            record["target_sha256"] if record["phase"] == "COMMITTED" else record["previous_sha256"]
-        )
-        if calculate_snapshot_sha256(self.adapter.read_current()) != expected_sha:
-            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
-        atomic_write_json(
-            self.journal.root / "current-snapshot.json",
-            record["target"] if record["phase"] == "COMMITTED" else record["previous"],
-        )
-        atomic_write_json(self.journal.root / "rollback-snapshot.json", record["previous"])
-        self.journal.write_receipt(record["operation_id"], record)
-        if not self.journal.audit.has_event(record["operation_id"], event):
-            payload = {"event": event}
-            if event == "SINGLE_OWNER_ACTIVATION_COMMITTED":
-                payload.update(
-                    payload_hash=record["payload_hash"],
-                    target_sha256=record["target_sha256"],
-                )
-            self.journal.audit.append(record["operation_id"], payload)
-        atomic_write_json(self.journal.root / "state.json", record)
-
-    def activate(self, bindings: Mapping[str, Any], supplied_plan_id: str) -> dict[str, Any]:
-        with exclusive_lock(self.lock_path):
-            self.journal.audit.verify()
-            validate_bindings(bindings)
-            operation_id = str(bindings["operation_id"])
-            payload_hash = binding_payload_hash(bindings)
-            record = self.journal.load(operation_id)
-            if record:
-                if record.get("payload_hash") != payload_hash:
-                    raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
-                if record.get("plan_id") != supplied_plan_id:
-                    raise RuntimeError("PLAN_ID_MISMATCH")
-                record = self._recover(record)
-                if record["phase"] == "COMMITTED":
-                    self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
-                    return {"status": "ALREADY_COMMITTED", **record}
-            else:
-                planned = self.plan(bindings)
-                if supplied_plan_id != planned["plan_id"]:
-                    raise RuntimeError("PLAN_ID_MISMATCH")
-            previous = self.adapter.read_current()
-            target = dict(bindings["target_branch_protection"])
-            if record is None:
-                record = {
-                    "task_id": bindings["task_id"],
-                    "operation_id": operation_id,
-                    "transaction_id": operation_id,
-                    "payload_hash": payload_hash,
-                    "plan_id": supplied_plan_id,
-                    "phase": "PREPARED",
-                    "previous": previous,
-                    "target": target,
-                    "previous_sha256": calculate_snapshot_sha256(previous),
-                    "target_sha256": calculate_snapshot_sha256(target),
-                    "activation_approval_merge_oid": bindings["activation_approval_merge_oid"],
-                    "evidence_closure_oid": bindings["evidence_closure_oid"],
-                    "owner_review_oid": bindings["owner_review_oid"],
-                }
-                self.journal.write(operation_id, record)
-                atomic_write_json(self.journal.root / "current-snapshot.json", previous)
-                atomic_write_json(self.journal.root / "rollback-snapshot.json", previous)
-            crash_at = bindings.get("crash_at")
-            if crash_at == "before_apply":
-                raise InjectedCrash("before_apply")
-            try:
-                current_sha = calculate_snapshot_sha256(self.adapter.read_current())
-                if current_sha == record["previous_sha256"]:
-                    applied = self.adapter.apply_atomic(record["previous_sha256"], record["target"])
-                    record["target"] = applied
-                    record["target_sha256"] = calculate_snapshot_sha256(applied)
-                elif current_sha != record["target_sha256"]:
-                    raise CASConflict("CAS_CONFLICT")
-                if crash_at == "during_apply":
-                    raise InjectedCrash("during_apply")
-                record["phase"] = "APPLIED"
-                self.journal.write(operation_id, record)
-                if crash_at == "after_apply_before_receipt":
-                    raise InjectedCrash("after_apply_before_receipt")
-                if not self.adapter.verify_applied(record["target_sha256"]):
-                    raise RuntimeError("POST_ACTIVATION_VERIFY_FAILED")
-                record["phase"] = "COMMITTED"
-                record["single_owner_policy_active"] = True
-                record["break_glass_active"] = False
-                self.journal.write(operation_id, record)
-                if crash_at == "after_commit_wal":
-                    raise InjectedCrash("after_commit_wal")
-                self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
-                return {"status": "ACTIVATED", **record}
-            except InjectedCrash:
-                raise
-            except Exception as exc:
-                failure = {**record, "phase": "FAILED", "error": type(exc).__name__}
-                current_sha = calculate_snapshot_sha256(self.adapter.read_current())
-                if current_sha == record["target_sha256"]:
-                    self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
-                    if not self.adapter.verify_rollback(record["previous_sha256"]):
-                        raise RuntimeError("ROLLBACK_VERIFY_FAILED") from exc
-                    failure["phase"] = "ROLLED_BACK"
-                self.journal.write(operation_id, failure)
-                self.journal.write_receipt(operation_id, failure)
-                self.journal.audit.append(operation_id, {"event": failure["phase"], "error": type(exc).__name__})
-                raise
-
-    def verify(self, bindings: Mapping[str, Any], operation_id: str) -> dict[str, Any]:
-        validate_bindings(bindings)
-        record = self.journal.load(operation_id)
-        if not record:
-            raise RuntimeError("TRANSACTION_NOT_FOUND")
-        if record.get("payload_hash") != binding_payload_hash(bindings):
-            raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
-        self.journal.audit.verify()
-        current_sha = calculate_snapshot_sha256(self.adapter.read_current())
-        passed = record.get("phase") == "COMMITTED" and current_sha == record.get("target_sha256")
-        if not passed:
-            raise RuntimeError("ACTIVATION_VERIFY_FAILED")
-        return {"status": "VERIFIED", "operation_id": operation_id, "target_sha256": current_sha}
-
-    def rollback(
-        self, bindings: Mapping[str, Any], operation_id: str, supplied_plan_id: str
-    ) -> dict[str, Any]:
-        with exclusive_lock(self.lock_path):
-            self.journal.audit.verify()
-            validate_bindings(bindings)
-            record = self.journal.load(operation_id)
-            if not record:
-                raise RuntimeError("TRANSACTION_NOT_FOUND")
-            if record.get("payload_hash") != binding_payload_hash(bindings):
-                raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
-            if record.get("plan_id") != supplied_plan_id:
-                raise RuntimeError("PLAN_ID_MISMATCH")
-            for field in (
-                "activation_approval_merge_oid",
-                "evidence_closure_oid",
-                "owner_review_oid",
-            ):
-                if str(record.get(field)) != str(bindings[field]):
-                    raise RuntimeError("ROLLBACK_AUTHORIZATION_MISMATCH_DENIED")
-            current_sha = calculate_snapshot_sha256(self.adapter.read_current())
-            if current_sha == record["previous_sha256"]:
-                record["phase"] = "ROLLED_BACK"
-                record["single_owner_policy_active"] = False
-                self.journal.write(operation_id, record)
-                self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
-                return {"status": "ALREADY_ROLLED_BACK", "operation_id": operation_id}
-            if current_sha != record["target_sha256"]:
-                raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
-            self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
-            if not self.adapter.verify_rollback(record["previous_sha256"]):
-                raise RuntimeError("ROLLBACK_VERIFY_FAILED")
-            if bindings.get("crash_at") == "after_rollback_cas":
-                raise InjectedCrash("after_rollback_cas")
-            record["phase"] = "ROLLED_BACK"
-            record["single_owner_policy_active"] = False
-            self.journal.write(operation_id, record)
-            if bindings.get("crash_at") == "after_rollback_wal":
-                raise InjectedCrash("after_rollback_wal")
-            self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
-            return {"status": "ROLLED_BACK", "operation_id": operation_id}
-
-    def status(self, operation_id: str | None = None) -> dict[str, Any]:
-        audit_head = self.journal.audit.verify()
-        if operation_id:
-            record = self.journal.load(operation_id)
-            if not record:
-                raise RuntimeError("TRANSACTION_NOT_FOUND")
-            return {"status": "OK", "audit_head": audit_head, "transaction": record}
-        state_path = self.journal.root / "state.json"
-        return {
-            "status": "OK",
-            "audit_head": audit_head,
-            "state": read_json(state_path) if state_path.exists() else None,
-            "transactions": list(self.journal.iter_transactions()),
-        }
-
-
-def load_bindings(path: Path) -> dict[str, Any]:
-    return read_json(path)
-
-
-def adapter_from_bindings(bindings: Mapping[str, Any]) -> Adapter:
-    backend = bindings.get("backend", "file")
-    backend_path = Path(str(bindings.get("backend_path", "")))
-    if not backend_path.is_absolute():
-        raise RuntimeError("BACKEND_PATH_MUST_BE_ABSOLUTE")
-    if backend == "file":
-        return FileProtectedBranchAdapter(backend_path)
-    if backend == "sqlite":
-        return SQLiteProtectedBranchAdapter(
-            backend_path,
-            repo_id=int(bindings.get("repo_id", 1)),
-            branch=str(bindings.get("branch", "governance/approvals")),
-            expected_schema_sha256=str(bindings["expected_gitea_schema_sha256"]),
-        )
-    raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
-
-
-def build_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser()
-    action = parser.add_mutually_exclusive_group(required=True)
-    for name in ("plan", "activate", "verify", "rollback", "status"):
-        action.add_argument(f"--{name}", action="store_true")
-    parser.add_argument("--bindings", type=Path)
-    parser.add_argument("--plan-id")
-    parser.add_argument("--transaction")
-    parser.add_argument("--state-root", type=Path, default=DEFAULT_STATE_ROOT)
-    parser.add_argument("--lock-path", type=Path, default=DEFAULT_LOCK_PATH)
-    return parser
-
-
-def main(argv: list[str] | None = None) -> int:
-    args = build_parser().parse_args(argv)
-    if args.status and args.bindings is None:
-        # Status without bindings is intentionally limited to durable local state.
-        journal = JournalStore(args.state_root)
-        result = {
-            "status": "OK",
-            "audit_head": journal.audit.verify(),
-            "transactions": list(journal.iter_transactions()),
-        }
-        print(json.dumps(result, sort_keys=True))
-        return 0
-    if args.bindings is None:
-        raise SystemExit("--bindings is required")
-    bindings = load_bindings(args.bindings)
-    engine = ActivationEngine(adapter_from_bindings(bindings), args.state_root, args.lock_path)
-    if args.plan:
-        result = engine.plan(bindings)
-    elif args.activate:
-        if not args.plan_id:
-            raise SystemExit("--plan-id is required")
-        result = engine.activate(bindings, args.plan_id)
-    elif args.verify:
-        result = engine.verify(bindings, args.transaction or str(bindings["operation_id"]))
-    elif args.rollback:
-        if not args.plan_id:
-            raise SystemExit("--plan-id is required")
-        result = engine.rollback(
-            bindings, args.transaction or str(bindings["operation_id"]), args.plan_id
-        )
-    else:
-        result = engine.status(args.transaction)
-    print(json.dumps(redact(result), sort_keys=True))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/evidence.py

@@ -1,61 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-import hashlib
-import json
-from typing import Mapping
-
-
-def canonical_json(value: object) -> bytes:
-    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
-
-
-@dataclass(frozen=True)
-class EvidenceClosure:
-    evidence: Mapping[str, str]
-    closure_sha256: str
-
-
-class ActivationEvidenceClosureBuilder:
-    def build(self, evidence: Mapping[str, str]) -> EvidenceClosure:
-        required = {
-            "activation_prerequisites_commit_oid",
-            "activation_prerequisites_tree_oid",
-            "artifact_sha256",
-            "rollback_artifact_sha256",
-            "validator_sha256",
-            "state_machine_sha256",
-            "branch_protection_adapter_sha256",
-            "rollback_configuration_sha256",
-            "test_evidence_sha256",
-            "deployment_manifest_sha256",
-            "dormant_unit_manifest_sha256",
-            "dormant_deployment_evidence_sha256",
-        }
-        if set(evidence) != required:
-            raise ValueError("evidence set mismatch")
-        for value in evidence.values():
-            if len(value) != 40 and len(value) != 64:
-                raise ValueError("invalid evidence digest")
-        return EvidenceClosure(
-            dict(evidence), hashlib.sha256(canonical_json(evidence)).hexdigest()
-        )
-
-
-class ApprovalMirrorStatusVerifier:
-    def verify(
-        self,
-        expected_oid: str,
-        *,
-        gitea_oid: str,
-        quarantine_oid: str,
-        authoritative_oid: str,
-        controller_visible_oid: str,
-        fsck_connectivity: str,
-    ) -> bool:
-        return (
-            len(expected_oid) == 40
-            and {gitea_oid, quarantine_oid, authoritative_oid, controller_visible_oid}
-            == {expected_oid}
-            and fsck_connectivity == "PASS"
-        )

services/aios-single-owner-governance/aios_single_owner_governance/fixture.py

@@ -1,26 +0,0 @@
-from __future__ import annotations
-
-from contextlib import AbstractContextManager
-from pathlib import Path
-import shutil
-import tempfile
-
-from .models import ProtectionSnapshot
-from .protection import MemoryProtectionStore
-
-
-class DisposableGiteaFixture(AbstractContextManager):
-    def __init__(self, initial: ProtectionSnapshot) -> None:
-        self.root = Path(tempfile.mkdtemp(prefix="aios-governance-fixture-"))
-        self.protection = MemoryProtectionStore(initial)
-        self.destroyed = False
-
-    def __enter__(self) -> "DisposableGiteaFixture":
-        return self
-
-    def __exit__(self, exc_type, exc, tb) -> None:
-        resolved = self.root.resolve()
-        if not resolved.name.startswith("aios-governance-fixture-"):
-            raise RuntimeError("fixture cleanup path guard failed")
-        shutil.rmtree(resolved)
-        self.destroyed = True

services/aios-single-owner-governance/aios_single_owner_governance/gitea_protected_branch.py

@@ -1,260 +0,0 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import sqlite3
-import time
-from typing import Any, Mapping
-
-from .journal import atomic_write_json, read_json, sha256_json
-
-
-class CASConflict(RuntimeError):
-    pass
-
-
-class UnsafePolicy(RuntimeError):
-    pass
-
-
-def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
-    return sha256_json(dict(snapshot))
-
-
-def validate_target(previous: Mapping[str, Any], target: Mapping[str, Any]) -> None:
-    if set(previous) != set(target):
-        raise UnsafePolicy("PROTECTION_ROW_SHAPE_CHANGED")
-    for field in ("id", "repo_id", "branch_name"):
-        if previous.get(field) != target.get(field):
-            raise UnsafePolicy(f"PROTECTION_IDENTITY_CHANGED:{field}")
-    if target.get("branch_name") != "governance/approvals":
-        raise UnsafePolicy("BRANCH_BINDING_DENIED")
-    if int(target.get("can_push", 1)) != 0:
-        raise UnsafePolicy("DIRECT_PUSH_MUST_REMAIN_DISABLED")
-    if int(target.get("required_approvals", 0)) != 1:
-        raise UnsafePolicy("EXACTLY_ONE_APPROVAL_REQUIRED")
-    if int(target.get("enable_approvals_whitelist", 0)) != 1:
-        raise UnsafePolicy("APPROVAL_WHITELIST_MUST_REMAIN_ENABLED")
-    if json.loads(target.get("approvals_whitelist_user_i_ds", "[]")) != [2]:
-        raise UnsafePolicy("OWNER_APPROVAL_WHITELIST_REQUIRED")
-    if int(target.get("enable_merge_whitelist", 0)) != 1:
-        raise UnsafePolicy("MERGE_WHITELIST_MUST_REMAIN_ENABLED")
-    if target.get("merge_whitelist_user_i_ds") != previous.get("merge_whitelist_user_i_ds"):
-        raise UnsafePolicy("MERGE_EXECUTOR_WHITELIST_CHANGE_DENIED")
-    if int(target.get("dismiss_stale_approvals", 0)) != 1:
-        raise UnsafePolicy("DISMISS_STALE_APPROVALS_REQUIRED")
-    if int(target.get("enable_status_check", 0)) != 1:
-        raise UnsafePolicy("STATUS_CHECKS_REQUIRED")
-    required_contexts = {
-        "aios/validator",
-        "aios/state-machine",
-        "aios/branch-protection",
-        "aios/secret-scan",
-        "aios/evidence-closure",
-    }
-    try:
-        contexts = set(json.loads(target.get("status_check_contexts", "[]")))
-    except (TypeError, ValueError) as exc:
-        raise UnsafePolicy("STATUS_CHECK_CONTEXTS_INVALID") from exc
-    if contexts != required_contexts:
-        raise UnsafePolicy("STATUS_CHECK_CONTEXTS_MISMATCH")
-    if int(target.get("block_on_rejected_reviews", 0)) != 1:
-        raise UnsafePolicy("REJECTED_REVIEWS_MUST_BLOCK")
-    if int(target.get("block_on_outdated_branch", 0)) != 1:
-        raise UnsafePolicy("OUTDATED_BRANCH_MUST_BLOCK")
-    if target.get("require_signed_commits") != previous.get("require_signed_commits"):
-        raise UnsafePolicy("SIGNATURE_POLICY_CHANGE_DENIED")
-
-
-class FileProtectedBranchAdapter:
-    def __init__(self, path: Path) -> None:
-        self.path = path
-
-    @staticmethod
-    def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
-        return calculate_snapshot_sha256(snapshot)
-
-    def read_current(self) -> dict[str, Any]:
-        return read_json(self.path)
-
-    def compare_expected(self, expected_sha256: str) -> bool:
-        return calculate_snapshot_sha256(self.read_current()) == expected_sha256
-
-    def apply_atomic(
-        self, expected_sha256: str, target: Mapping[str, Any]
-    ) -> dict[str, Any]:
-        previous = self.read_current()
-        if calculate_snapshot_sha256(previous) != expected_sha256:
-            raise CASConflict("CAS_CONFLICT")
-        validate_target(previous, target)
-        atomic_write_json(self.path, dict(target))
-        applied = self.read_current()
-        if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(target):
-            raise CASConflict("CAS_VERIFY_FAILED")
-        return applied
-
-    def verify_applied(self, expected_sha256: str) -> bool:
-        return self.compare_expected(expected_sha256)
-
-    def rollback_atomic(
-        self, expected_target_sha256: str, previous: Mapping[str, Any]
-    ) -> dict[str, Any]:
-        current = self.read_current()
-        if calculate_snapshot_sha256(current) != expected_target_sha256:
-            raise CASConflict("ROLLBACK_CONFLICT")
-        atomic_write_json(self.path, dict(previous))
-        restored = self.read_current()
-        if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
-            raise CASConflict("ROLLBACK_VERIFY_FAILED")
-        return restored
-
-    def verify_rollback(self, expected_sha256: str) -> bool:
-        return self.compare_expected(expected_sha256)
-
-
-class SQLiteProtectedBranchAdapter:
-    ALLOWED_UPDATE_FIELDS = (
-        "can_push",
-        "enable_merge_whitelist",
-        "merge_whitelist_user_i_ds",
-        "enable_status_check",
-        "status_check_contexts",
-        "enable_approvals_whitelist",
-        "approvals_whitelist_user_i_ds",
-        "required_approvals",
-        "block_on_rejected_reviews",
-        "block_on_official_review_requests",
-        "block_on_outdated_branch",
-        "dismiss_stale_approvals",
-        "ignore_stale_approvals",
-        "updated_unix",
-    )
-
-    def __init__(
-        self,
-        db_path: Path,
-        *,
-        repo_id: int = 1,
-        branch: str = "governance/approvals",
-        expected_schema_sha256: str,
-    ) -> None:
-        self.db_path = db_path
-        self.repo_id = repo_id
-        self.branch = branch
-        self.expected_schema_sha256 = expected_schema_sha256
-
-    def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self.db_path, timeout=10)
-        connection.row_factory = sqlite3.Row
-        return connection
-
-    def schema_sha256(self, connection: sqlite3.Connection | None = None) -> str:
-        own = connection is None
-        connection = connection or self._connect()
-        try:
-            rows = [dict(row) for row in connection.execute("PRAGMA table_info(protected_branch)")]
-            return sha256_json(rows)
-        finally:
-            if own:
-                connection.close()
-
-    def _check_schema(self, connection: sqlite3.Connection) -> None:
-        if not isinstance(self.expected_schema_sha256, str) or len(self.expected_schema_sha256) != 64:
-            raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_REQUIRED")
-        if self.schema_sha256(connection) != self.expected_schema_sha256:
-            raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_MISMATCH")
-
-    def _read(self, connection: sqlite3.Connection) -> dict[str, Any]:
-        row = connection.execute(
-            "SELECT * FROM protected_branch WHERE repo_id=? AND branch_name=?",
-            (self.repo_id, self.branch),
-        ).fetchone()
-        if row is None:
-            raise RuntimeError("PROTECTED_BRANCH_ROW_NOT_FOUND")
-        return dict(row)
-
-    def read_current(self) -> dict[str, Any]:
-        with self._connect() as connection:
-            self._check_schema(connection)
-            return self._read(connection)
-
-    @staticmethod
-    def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
-        return calculate_snapshot_sha256(snapshot)
-
-    def compare_expected(self, expected_sha256: str) -> bool:
-        return calculate_snapshot_sha256(self.read_current()) == expected_sha256
-
-    def apply_atomic(
-        self, expected_sha256: str, target: Mapping[str, Any]
-    ) -> dict[str, Any]:
-        connection = self._connect()
-        try:
-            connection.execute("BEGIN IMMEDIATE")
-            self._check_schema(connection)
-            previous = self._read(connection)
-            if calculate_snapshot_sha256(previous) != expected_sha256:
-                raise CASConflict("CAS_CONFLICT")
-            validate_target(previous, target)
-            values = dict(target)
-            values["updated_unix"] = int(values.get("updated_unix") or time.time())
-            assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
-            parameters = [values[field] for field in self.ALLOWED_UPDATE_FIELDS]
-            parameters.extend((self.repo_id, self.branch))
-            cursor = connection.execute(
-                f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
-                parameters,
-            )
-            if cursor.rowcount != 1:
-                raise CASConflict("CAS_ROWCOUNT_INVALID")
-            applied = self._read(connection)
-            expected_target = dict(target)
-            expected_target["updated_unix"] = values["updated_unix"]
-            if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(expected_target):
-                raise CASConflict("CAS_VERIFY_FAILED")
-            connection.commit()
-            return applied
-        except Exception:
-            connection.rollback()
-            raise
-        finally:
-            connection.close()
-
-    def verify_applied(self, expected_sha256: str) -> bool:
-        return self.compare_expected(expected_sha256)
-
-    def rollback_atomic(
-        self, expected_target_sha256: str, previous: Mapping[str, Any]
-    ) -> dict[str, Any]:
-        connection = self._connect()
-        try:
-            connection.execute("BEGIN IMMEDIATE")
-            self._check_schema(connection)
-            current = self._read(connection)
-            if calculate_snapshot_sha256(current) != expected_target_sha256:
-                raise CASConflict("ROLLBACK_CONFLICT")
-            for field in ("id", "repo_id", "branch_name"):
-                if current.get(field) != previous.get(field):
-                    raise CASConflict("ROLLBACK_IDENTITY_MISMATCH")
-            assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
-            parameters = [previous[field] for field in self.ALLOWED_UPDATE_FIELDS]
-            parameters.extend((self.repo_id, self.branch))
-            cursor = connection.execute(
-                f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
-                parameters,
-            )
-            if cursor.rowcount != 1:
-                raise CASConflict("ROLLBACK_ROWCOUNT_INVALID")
-            restored = self._read(connection)
-            if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
-                raise CASConflict("ROLLBACK_VERIFY_FAILED")
-            connection.commit()
-            return restored
-        except Exception:
-            connection.rollback()
-            raise
-        finally:
-            connection.close()
-
-    def verify_rollback(self, expected_sha256: str) -> bool:
-        return self.compare_expected(expected_sha256)

services/aios-single-owner-governance/aios_single_owner_governance/journal.py

@@ -1,189 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-from pathlib import Path
-import tempfile
-from typing import Any, Iterable
-
-
-SENSITIVE_FRAGMENTS = (
-    "authorization",
-    "cookie",
-    "credential",
-    "password",
-    "private_key",
-    "secret",
-    "token",
-)
-
-
-def canonical_json(value: object) -> bytes:
-    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
-
-
-def sha256_json(value: object) -> str:
-    return hashlib.sha256(canonical_json(value)).hexdigest()
-
-
-def redact(value: Any) -> Any:
-    if isinstance(value, dict):
-        return {
-            str(key): (
-                "[REDACTED]"
-                if any(fragment in str(key).lower() for fragment in SENSITIVE_FRAGMENTS)
-                else redact(item)
-            )
-            for key, item in value.items()
-        }
-    if isinstance(value, list):
-        return [redact(item) for item in value]
-    if isinstance(value, tuple):
-        return [redact(item) for item in value]
-    return value
-
-
-def ensure_state_directory(path: Path) -> None:
-    if path.is_symlink():
-        raise RuntimeError("STATE_DIRECTORY_SYMLINK_DENIED")
-    path.mkdir(parents=True, exist_ok=True, mode=0o750)
-    path.chmod(0o750)
-    mode = path.stat().st_mode & 0o777
-    if mode != 0o750:
-        raise RuntimeError("STATE_DIRECTORY_MODE_INVALID")
-
-
-def atomic_write_json(path: Path, value: object) -> None:
-    ensure_state_directory(path.parent)
-    raw = json.dumps(redact(value), sort_keys=True, indent=2).encode("utf-8") + b"\n"
-    with tempfile.NamedTemporaryFile(
-        dir=path.parent, prefix=f".{path.name}.", delete=False
-    ) as handle:
-        temporary = Path(handle.name)
-        handle.write(raw)
-        handle.flush()
-        os.fsync(handle.fileno())
-    temporary.chmod(0o640)
-    os.replace(temporary, path)
-    path.chmod(0o640)
-    directory_fd = os.open(path.parent, os.O_RDONLY)
-    try:
-        os.fsync(directory_fd)
-    finally:
-        os.close(directory_fd)
-
-
-def read_json(path: Path) -> dict[str, Any]:
-    raw = json.loads(path.read_text(encoding="utf-8"))
-    if not isinstance(raw, dict):
-        raise RuntimeError("STATE_DOCUMENT_NOT_OBJECT")
-    return raw
-
-
-class AuditChain:
-    def __init__(self, path: Path) -> None:
-        self.path = path
-
-    def _records(self) -> list[dict[str, Any]]:
-        if not self.path.exists():
-            return []
-        records: list[dict[str, Any]] = []
-        for line in self.path.read_text(encoding="utf-8").splitlines():
-            if line.strip():
-                record = json.loads(line)
-                if not isinstance(record, dict):
-                    raise RuntimeError("AUDIT_RECORD_INVALID")
-                records.append(record)
-        return records
-
-    def verify(self) -> str:
-        previous = "0" * 64
-        for index, record in enumerate(self._records(), 1):
-            payload = record.get("payload")
-            if record.get("sequence") != index:
-                raise RuntimeError("AUDIT_SEQUENCE_INVALID")
-            if record.get("previous_event_hash") != previous:
-                raise RuntimeError("AUDIT_PREVIOUS_HASH_INVALID")
-            payload_hash = sha256_json(payload)
-            if record.get("payload_hash") != payload_hash:
-                raise RuntimeError("AUDIT_PAYLOAD_HASH_INVALID")
-            material = {
-                "sequence": index,
-                "previous_event_hash": previous,
-                "payload_hash": payload_hash,
-                "transaction_id": record.get("transaction_id"),
-            }
-            event_hash = sha256_json(material)
-            if record.get("event_hash") != event_hash:
-                raise RuntimeError("AUDIT_EVENT_HASH_INVALID")
-            previous = event_hash
-        return previous
-
-    def append(self, transaction_id: str, payload: dict[str, Any]) -> dict[str, Any]:
-        ensure_state_directory(self.path.parent)
-        previous = self.verify()
-        sequence = len(self._records()) + 1
-        clean_payload = redact(payload)
-        payload_hash = sha256_json(clean_payload)
-        material = {
-            "sequence": sequence,
-            "previous_event_hash": previous,
-            "payload_hash": payload_hash,
-            "transaction_id": transaction_id,
-        }
-        record = {
-            **material,
-            "payload": clean_payload,
-            "event_hash": sha256_json(material),
-        }
-        fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o640)
-        try:
-            os.write(fd, canonical_json(record) + b"\n")
-            os.fsync(fd)
-        finally:
-            os.close(fd)
-        self.path.chmod(0o640)
-        return record
-
-    def has_event(self, transaction_id: str, event: str) -> bool:
-        self.verify()
-        return any(
-            record.get("transaction_id") == transaction_id
-            and isinstance(record.get("payload"), dict)
-            and record["payload"].get("event") == event
-            for record in self._records()
-        )
-
-
-class JournalStore:
-    def __init__(self, root: Path) -> None:
-        self.root = root
-        self.wal = root / "wal"
-        self.receipts = root / "receipts"
-        ensure_state_directory(root)
-        ensure_state_directory(self.wal)
-        ensure_state_directory(self.receipts)
-        self.audit = AuditChain(root / "audit.jsonl")
-
-    def transaction_path(self, operation_id: str) -> Path:
-        if not operation_id or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for char in operation_id):
-            raise RuntimeError("OPERATION_ID_INVALID")
-        return self.wal / f"{operation_id}.json"
-
-    def receipt_path(self, operation_id: str) -> Path:
-        return self.receipts / f"{operation_id}.json"
-
-    def load(self, operation_id: str) -> dict[str, Any] | None:
-        path = self.transaction_path(operation_id)
-        return read_json(path) if path.exists() else None
-
-    def write(self, operation_id: str, value: dict[str, Any]) -> None:
-        atomic_write_json(self.transaction_path(operation_id), value)
-
-    def write_receipt(self, operation_id: str, value: dict[str, Any]) -> None:
-        atomic_write_json(self.receipt_path(operation_id), value)
-
-    def iter_transactions(self) -> Iterable[dict[str, Any]]:
-        for path in sorted(self.wal.glob("*.json")):
-            yield read_json(path)

services/aios-single-owner-governance/aios_single_owner_governance/lock.py

@@ -1,47 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-try:
-    import fcntl
-except ImportError:  # pragma: no cover - exercised by Windows validation
-    fcntl = None
-    import msvcrt
-
-
-class MaintenanceLock:
-    """Process fence used only by fixture and later approved activation executors."""
-
-    def __init__(self, path: Path) -> None:
-        self.path = path
-        self._handle = None
-
-    def __enter__(self) -> "MaintenanceLock":
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self._handle = self.path.open("a+")
-        if fcntl is not None:
-            try:
-                fcntl.flock(self._handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-            except BlockingIOError:
-                self._handle.close()
-                self._handle = None
-                raise
-        else:
-            self._handle.seek(0)
-            try:
-                msvcrt.locking(self._handle.fileno(), msvcrt.LK_NBLCK, 1)
-            except OSError as exc:
-                self._handle.close()
-                self._handle = None
-                raise BlockingIOError("maintenance lock held") from exc
-        return self
-
-    def __exit__(self, exc_type, exc, tb) -> None:
-        if self._handle is not None:
-            if fcntl is not None:
-                fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN)
-            else:
-                self._handle.seek(0)
-                msvcrt.locking(self._handle.fileno(), msvcrt.LK_UNLCK, 1)
-            self._handle.close()
-            self._handle = None

services/aios-single-owner-governance/aios_single_owner_governance/models.py

@@ -1,90 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from enum import Enum
-from typing import Any, Mapping
-
-
-class GovernanceState(str, Enum):
-    HUMAN_OWNER_APPROVAL_PENDING = "HUMAN_OWNER_APPROVAL_PENDING"
-    OWNER_APPROVAL_VERIFIED = "OWNER_APPROVAL_VERIFIED"
-    MACHINE_GATES_VERIFIED = "MACHINE_GATES_VERIFIED"
-    AUTO_MERGE_READY = "AUTO_MERGE_READY"
-    OWNER_APPROVED_EXECUTION_READY = "OWNER_APPROVED_EXECUTION_READY"
-
-
-@dataclass(frozen=True)
-class Review:
-    reviewer: str
-    reviewer_id: int
-    state: str
-    head_oid: str
-    reviewed_at: datetime
-    stale: bool = False
-    dismissed: bool = False
-
-    def __post_init__(self) -> None:
-        if self.reviewed_at.tzinfo is None:
-            raise ValueError("reviewed_at must be timezone-aware")
-
-
-@dataclass(frozen=True)
-class MachineGate:
-    context: str
-    state: str
-    head_oid: str
-    observed_at: datetime
-
-    def __post_init__(self) -> None:
-        if self.observed_at.tzinfo is None:
-            raise ValueError("observed_at must be timezone-aware")
-
-
-@dataclass(frozen=True)
-class ApprovalContext:
-    current_head_oid: str
-    requester: str
-    executor: str
-    owner: str
-    owner_id: int
-    reviews: tuple[Review, ...]
-    machine_gates: tuple[MachineGate, ...]
-    required_machine_contexts: tuple[str, ...]
-    expires_at: datetime
-    wip: bool
-    now: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-
-    def __post_init__(self) -> None:
-        if self.expires_at.tzinfo is None or self.now.tzinfo is None:
-            raise ValueError("approval timestamps must be timezone-aware")
-
-
-@dataclass(frozen=True)
-class ProtectionSnapshot:
-    branch: str
-    required_approvals: int
-    approval_whitelist: tuple[str, ...]
-    merge_whitelist: tuple[str, ...]
-    dismiss_stale_approvals: bool
-    require_status_checks: bool
-    status_contexts: tuple[str, ...]
-    direct_push: bool
-    force_push: bool
-    branch_deletion: bool
-    version: int
-
-    def as_dict(self) -> Mapping[str, Any]:
-        return {
-            "branch": self.branch,
-            "required_approvals": self.required_approvals,
-            "approval_whitelist": list(self.approval_whitelist),
-            "merge_whitelist": list(self.merge_whitelist),
-            "dismiss_stale_approvals": self.dismiss_stale_approvals,
-            "require_status_checks": self.require_status_checks,
-            "status_contexts": list(self.status_contexts),
-            "direct_push": self.direct_push,
-            "force_push": self.force_push,
-            "branch_deletion": self.branch_deletion,
-            "version": self.version,
-        }

services/aios-single-owner-governance/aios_single_owner_governance/policy.py

@@ -1,23 +0,0 @@
-from __future__ import annotations
-
-
-class OperationDenied(PermissionError):
-    pass
-
-
-class DormantOperationGuard:
-    """Fail-closed operation boundary for the prerequisite deployment."""
-
-    DENIED = frozenset({
-        "protected_branch.direct_push",
-        "governance.direct_push",
-        "tag.direct_write",
-        "git.force_push",
-        "production.without_owner",
-        "branch_protection.live_update",
-        "single_owner_policy.activate",
-    })
-
-    def authorize(self, operation: str, *, fixture: bool = False) -> None:
-        if operation in self.DENIED or not fixture:
-            raise OperationDenied(operation)

services/aios-single-owner-governance/aios_single_owner_governance/protection.py

@@ -1,98 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-import hashlib
-import json
-from typing import Protocol
-
-from .models import ProtectionSnapshot
-
-
-class CASConflict(RuntimeError):
-    pass
-
-
-class DormantViolation(RuntimeError):
-    pass
-
-
-def snapshot_sha256(snapshot: ProtectionSnapshot) -> str:
-    raw = json.dumps(snapshot.as_dict(), sort_keys=True, separators=(",", ":")).encode()
-    return hashlib.sha256(raw).hexdigest()
-
-
-class ProtectionStore(Protocol):
-    fixture: bool
-
-    def read(self, branch: str) -> ProtectionSnapshot: ...
-    def compare_and_swap(
-        self, branch: str, expected_sha256: str, target: ProtectionSnapshot
-    ) -> bool: ...
-
-
-@dataclass(frozen=True)
-class ProtectionPlan:
-    branch: str
-    expected_previous_sha256: str
-    previous: ProtectionSnapshot
-    target: ProtectionSnapshot
-    target_sha256: str
-
-
-class BranchProtectionCASMigrationAdapter:
-    """Candidate adapter. Mutations are restricted to disposable fixtures."""
-
-    def plan(
-        self,
-        current: ProtectionSnapshot,
-        target: ProtectionSnapshot,
-        expected_previous_sha256: str,
-    ) -> ProtectionPlan:
-        actual = snapshot_sha256(current)
-        if actual != expected_previous_sha256:
-            raise CASConflict("EXPECTED_PREVIOUS_SHA256_MISMATCH")
-        if current.branch != target.branch:
-            raise CASConflict("BRANCH_BINDING_MISMATCH")
-        if target.direct_push or target.force_push or target.branch_deletion:
-            raise DormantViolation("UNSAFE_TARGET_POLICY")
-        return ProtectionPlan(
-            current.branch, actual, current, target, snapshot_sha256(target)
-        )
-
-    def apply_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
-        if not store.fixture:
-            raise DormantViolation("LIVE_PROTECTION_MUTATION_DENIED")
-        if not store.compare_and_swap(
-            plan.branch, plan.expected_previous_sha256, plan.target
-        ):
-            raise CASConflict("CONCURRENT_PROTECTION_CHANGE")
-        return store.read(plan.branch)
-
-    def rollback_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
-        if not store.fixture:
-            raise DormantViolation("LIVE_PROTECTION_ROLLBACK_DENIED")
-        if not store.compare_and_swap(plan.branch, plan.target_sha256, plan.previous):
-            raise CASConflict("ROLLBACK_CONFLICT")
-        return store.read(plan.branch)
-
-
-class MemoryProtectionStore:
-    fixture = True
-
-    def __init__(self, snapshot: ProtectionSnapshot) -> None:
-        self._snapshot = snapshot
-
-    def read(self, branch: str) -> ProtectionSnapshot:
-        if self._snapshot.branch != branch:
-            raise KeyError(branch)
-        return self._snapshot
-
-    def compare_and_swap(
-        self, branch: str, expected_sha256: str, target: ProtectionSnapshot
-    ) -> bool:
-        if self._snapshot.branch != branch:
-            return False
-        if snapshot_sha256(self._snapshot) != expected_sha256:
-            return False
-        self._snapshot = target
-        return True

services/aios-single-owner-governance/aios_single_owner_governance/rollback.py

@@ -1,69 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-import hashlib
-import json
-from pathlib import Path
-import tempfile
-
-from .models import ProtectionSnapshot
-from .protection import MemoryProtectionStore, ProtectionPlan, snapshot_sha256
-
-
-@dataclass(frozen=True)
-class RollbackConfiguration:
-    branch: str
-    previous_sha256: str
-    target_sha256: str
-    payload: dict
-
-
-class RollbackConfigurationManager:
-    def build(self, plan: ProtectionPlan) -> RollbackConfiguration:
-        return RollbackConfiguration(
-            plan.branch,
-            plan.expected_previous_sha256,
-            plan.target_sha256,
-            {"previous": plan.previous.as_dict(), "target": plan.target.as_dict()},
-        )
-
-    def install_fixture(self, root: Path, config: RollbackConfiguration) -> Path:
-        if not root.name.startswith("aios-governance-fixture-"):
-            raise PermissionError("rollback configuration may only be installed in fixture root")
-        root.mkdir(parents=True, exist_ok=True)
-        destination = root / "rollback-configuration.json"
-        raw = json.dumps(config.payload, sort_keys=True, indent=2).encode() + b"\n"
-        with tempfile.NamedTemporaryFile(dir=root, prefix=".rollback.", delete=False) as handle:
-            handle.write(raw)
-            handle.flush()
-            temporary = Path(handle.name)
-        temporary.chmod(0o600)
-        temporary.replace(destination)
-        return destination
-
-    def verify(self, config: RollbackConfiguration, plan: ProtectionPlan) -> bool:
-        return (
-            config.previous_sha256 == snapshot_sha256(plan.previous)
-            and config.target_sha256 == snapshot_sha256(plan.target)
-        )
-
-
-@dataclass
-class ProtectionCompensationSaga:
-    adapter: object
-    store: MemoryProtectionStore
-    plan: ProtectionPlan
-    applied: bool = False
-    rolled_back: bool = False
-
-    def execute_fixture(self) -> ProtectionSnapshot:
-        result = self.adapter.apply_fixture(self.store, self.plan)
-        self.applied = True
-        return result
-
-    def compensate_fixture(self) -> ProtectionSnapshot:
-        if not self.applied:
-            return self.store.read(self.plan.branch)
-        result = self.adapter.rollback_fixture(self.store, self.plan)
-        self.rolled_back = True
-        return result

services/aios-single-owner-governance/aios_single_owner_governance/service.py

@@ -1,69 +0,0 @@
-from __future__ import annotations
-
-import argparse
-import json
-import os
-from pathlib import Path
-import tempfile
-
-
-def runtime_state(component: str, prior_invocations: int = 0) -> dict:
-    return {
-        "component": component,
-        "deployment_mode": "DORMANT_SHADOW_ONLY",
-        "enabled": False,
-        "active": False,
-        "production_routing": False,
-        "production_mutation": False,
-        "global_gitops_writes_enabled": False,
-        "prior_invocations": prior_invocations,
-        "restart_recovery": prior_invocations > 0,
-    }
-
-
-def checkpoint_state(component: str, root: Path) -> dict:
-    root.mkdir(parents=True, exist_ok=True)
-    destination = root / f"{component}.json"
-    prior = 0
-    if destination.exists():
-        previous = json.loads(destination.read_text(encoding="utf-8"))
-        if previous.get("component") != component or previous.get("deployment_mode") != "DORMANT_SHADOW_ONLY":
-            raise RuntimeError("checkpoint identity mismatch")
-        prior = int(previous.get("invocations", 0))
-    state = runtime_state(component, prior)
-    checkpoint = {**state, "invocations": prior + 1}
-    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=root, delete=False) as handle:
-        json.dump(checkpoint, handle, sort_keys=True)
-        handle.write("\n")
-        handle.flush()
-        temporary = Path(handle.name)
-    temporary.replace(destination)
-    return state
-
-
-def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--component", required=True, choices=("owner-review-adapter", "machine-status-producer"))
-    parser.add_argument("--self-check", action="store_true")
-    parser.add_argument("--activate", action="store_true")
-    parser.add_argument("--checkpoint-dir", type=Path)
-    args = parser.parse_args(argv)
-    if args.activate:
-        parser.error("activation is not authorized by the prerequisites candidate")
-    if os.environ.get("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED", "false").lower() != "false":
-        parser.error("candidate must remain disabled")
-    if os.environ.get("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE", "false").lower() != "false":
-        parser.error("activation must remain inactive")
-    if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
-        parser.error("production mutation must remain disabled")
-    state = (
-        checkpoint_state(args.component, args.checkpoint_dir)
-        if args.checkpoint_dir is not None
-        else runtime_state(args.component)
-    )
-    print(json.dumps(state, sort_keys=True))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/state_machine.py

@@ -1,54 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-import json
-from pathlib import Path
-import tempfile
-
-from .models import GovernanceState
-
-
-class InvalidTransition(RuntimeError):
-    pass
-
-
-_NEXT = {
-    GovernanceState.HUMAN_OWNER_APPROVAL_PENDING: GovernanceState.OWNER_APPROVAL_VERIFIED,
-    GovernanceState.OWNER_APPROVAL_VERIFIED: GovernanceState.MACHINE_GATES_VERIFIED,
-    GovernanceState.MACHINE_GATES_VERIFIED: GovernanceState.AUTO_MERGE_READY,
-    GovernanceState.AUTO_MERGE_READY: GovernanceState.OWNER_APPROVED_EXECUTION_READY,
-}
-
-
-@dataclass
-class GovernanceStateMachine:
-    state: GovernanceState = GovernanceState.HUMAN_OWNER_APPROVAL_PENDING
-
-    def transition(self, target: GovernanceState) -> GovernanceState:
-        if _NEXT.get(self.state) != target:
-            raise InvalidTransition(f"{self.state.value}->{target.value}")
-        self.state = target
-        return self.state
-
-    def restore(self, checkpoint: GovernanceState) -> None:
-        if checkpoint not in GovernanceState:
-            raise InvalidTransition("unknown checkpoint")
-        self.state = checkpoint
-
-    def save_fixture(self, path: Path) -> None:
-        if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
-            raise PermissionError("state checkpoint writes are fixture-only")
-        path.parent.mkdir(parents=True, exist_ok=True)
-        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
-            json.dump({"state": self.state.value}, handle, sort_keys=True)
-            handle.write("\n")
-            handle.flush()
-            temporary = Path(handle.name)
-        temporary.replace(path)
-
-    @classmethod
-    def load_fixture(cls, path: Path) -> "GovernanceStateMachine":
-        if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
-            raise PermissionError("state checkpoint reads are fixture-only")
-        raw = json.loads(path.read_text(encoding="utf-8"))
-        return cls(GovernanceState(raw["state"]))

services/aios-single-owner-governance/aios_single_owner_governance/status.py

@@ -1,38 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from datetime import datetime, timezone
-
-
-class DormantStatusProducer(RuntimeError):
-    pass
-
-
-@dataclass(frozen=True)
-class StatusRecord:
-    context: str
-    state: str
-    head_oid: str
-    observed_at: str
-
-
-class MachineStatusProducer:
-    def __init__(self, *, enabled: bool = False, fixture: bool = False) -> None:
-        self.enabled = enabled
-        self.fixture = fixture
-
-    def produce(self, context: str, state: str, head_oid: str) -> StatusRecord:
-        if not self.fixture:
-            raise DormantStatusProducer("LIVE_STATUS_PUBLICATION_DENIED")
-        if self.enabled:
-            raise DormantStatusProducer("DORMANT_CANDIDATE_MUST_NOT_BE_ENABLED")
-        if context not in {
-            "aios/machine-gates", "aios/secret-scan", "aios/evidence-closure"
-        }:
-            raise ValueError("unknown status context")
-        if state not in {"success", "failure"} or len(head_oid) != 40:
-            raise ValueError("invalid status record")
-        return StatusRecord(
-            context, state, head_oid,
-            datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
-        )

services/aios-single-owner-governance/aios_single_owner_governance/validator.py

@@ -1,69 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-
-from .models import ApprovalContext, Review
-
-
-class ValidationDenied(RuntimeError):
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-@dataclass(frozen=True)
-class ValidationResult:
-    owner_review: Review
-    machine_contexts: tuple[str, ...]
-
-
-class SingleOwnerApprovalValidator:
-    """Pure candidate validator. It has no Gitea or production write path."""
-
-    def validate(self, context: ApprovalContext) -> ValidationResult:
-        if context.now >= context.expires_at:
-            raise ValidationDenied("APPROVAL_EXPIRED")
-        if context.wip:
-            raise ValidationDenied("WIP_BLOCKED")
-        if context.requester == context.owner:
-            raise ValidationDenied("REQUESTER_SELF_APPROVAL")
-        if context.executor == context.owner:
-            raise ValidationDenied("EXECUTOR_SELF_APPROVAL")
-
-        approved = [review for review in context.reviews if review.state == "APPROVED"]
-        owner_reviews = [
-            review for review in approved
-            if review.reviewer == context.owner and review.reviewer_id == context.owner_id
-        ]
-        if len(owner_reviews) != 1:
-            raise ValidationDenied("OWNER_APPROVAL_REQUIRED")
-        owner_review = owner_reviews[0]
-        if owner_review.stale:
-            raise ValidationDenied("STALE_OWNER_APPROVAL")
-        if owner_review.dismissed:
-            raise ValidationDenied("DISMISSED_OWNER_APPROVAL")
-        if owner_review.head_oid != context.current_head_oid:
-            raise ValidationDenied("OWNER_APPROVAL_NOT_LATEST_HEAD")
-        if owner_review.reviewed_at >= context.expires_at:
-            raise ValidationDenied("OWNER_APPROVAL_EXPIRED")
-        if any(
-            review.state == "APPROVED"
-            and review.reviewer != context.owner
-            and review.reviewer_id != context.owner_id
-            for review in context.reviews
-        ):
-            # Non-owner approvals may exist for legacy mechanics, but never
-            # satisfy or replace the human authorization decision.
-            pass
-
-        gates = {gate.context: gate for gate in context.machine_gates}
-        if set(gates) != set(context.required_machine_contexts):
-            raise ValidationDenied("MACHINE_GATE_SET_MISMATCH")
-        for gate in gates.values():
-            if gate.state != "success":
-                raise ValidationDenied("MACHINE_GATE_FAILED")
-            if gate.head_oid != context.current_head_oid:
-                raise ValidationDenied("MACHINE_GATE_NOT_LATEST_HEAD")
-            if gate.observed_at >= context.expires_at:
-                raise ValidationDenied("MACHINE_GATE_EXPIRED")
-        return ValidationResult(owner_review, tuple(sorted(gates)))

services/aios-single-owner-governance/pyproject.toml

@@ -1,16 +0,0 @@
-[build-system]
-requires = ["setuptools>=68"]
-build-backend = "setuptools.build_meta"
-
-[project]
-name = "aios-single-owner-governance"
-version = "0.1.0"
-requires-python = ">=3.11"
-dependencies = []
-
-[project.scripts]
-aios-single-owner-governance = "aios_single_owner_governance.activation:main"
-
-[tool.setuptools.packages.find]
-where = ["."]
-include = ["aios_single_owner_governance*"]

tests/single_owner_governance/__init__.py

@@ -1 +0,0 @@
-

tests/single_owner_governance/helpers.py

@@ -1,51 +0,0 @@
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-
-from aios_single_owner_governance.models import (
-    ApprovalContext,
-    MachineGate,
-    ProtectionSnapshot,
-    Review,
-)
-
-HEAD = "a" * 40
-NOW = datetime(2026, 7, 28, 19, 0, tzinfo=timezone.utc)
-CONTEXTS = ("aios/machine-gates", "aios/secret-scan", "aios/evidence-closure")
-
-
-def owner_review(**changes) -> Review:
-    values = dict(
-        reviewer="owner", reviewer_id=2, state="APPROVED", head_oid=HEAD,
-        reviewed_at=NOW - timedelta(minutes=1), stale=False, dismissed=False,
-    )
-    values.update(changes)
-    return Review(**values)
-
-
-def gates(*, state="success", head_oid=HEAD):
-    return tuple(MachineGate(name, state, head_oid, NOW) for name in CONTEXTS)
-
-
-def context(**changes) -> ApprovalContext:
-    values = dict(
-        current_head_oid=HEAD, requester="agent", executor="merge-executor",
-        owner="owner", owner_id=2, reviews=(owner_review(),), machine_gates=gates(),
-        required_machine_contexts=CONTEXTS, expires_at=NOW + timedelta(hours=1),
-        wip=False, now=NOW,
-    )
-    values.update(changes)
-    return ApprovalContext(**values)
-
-
-def legacy_protection(version=1) -> ProtectionSnapshot:
-    return ProtectionSnapshot(
-        "governance/approvals", 2, ("codex", "gpt"), ("codex", "gpt"),
-        True, False, (), False, False, False, version,
-    )
-
-def single_owner_protection(version=2) -> ProtectionSnapshot:
-    return ProtectionSnapshot(
-        "governance/approvals", 1, ("owner",), ("owner",), True, True,
-        CONTEXTS, False, False, False, version,
-    )

tests/single_owner_governance/test_dormant_invariants.py

@@ -1,72 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-import unittest
-
-from aios_single_owner_governance.fixture import DisposableGiteaFixture
-from aios_single_owner_governance.protection import (
-    BranchProtectionCASMigrationAdapter,
-    DormantViolation,
-    snapshot_sha256,
-)
-from aios_single_owner_governance.policy import DormantOperationGuard, OperationDenied
-
-from .helpers import legacy_protection, single_owner_protection
-
-
-class LiveStore:
-    fixture = False
-
-    def read(self, branch):
-        return legacy_protection()
-
-    def compare_and_swap(self, branch, expected_sha256, target):
-        raise AssertionError("live compare_and_swap must never be invoked")
-
-
-class DormantInvariantTests(unittest.TestCase):
-    def test_live_protection_apply_and_rollback_are_denied(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old, new = legacy_protection(), single_owner_protection()
-        plan = adapter.plan(old, new, snapshot_sha256(old))
-        with self.assertRaises(DormantViolation):
-            adapter.apply_fixture(LiveStore(), plan)
-        with self.assertRaises(DormantViolation):
-            adapter.rollback_fixture(LiveStore(), plan)
-
-    def test_unsafe_policy_cannot_be_planned(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old = legacy_protection()
-        for unsafe in (
-            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "direct_push": True}),
-            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "force_push": True}),
-            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "branch_deletion": True}),
-        ):
-            with self.assertRaises(DormantViolation):
-                adapter.plan(old, unsafe, snapshot_sha256(old))
-
-    def test_disposable_fixture_is_destroyed(self):
-        fixture = DisposableGiteaFixture(legacy_protection())
-        root = fixture.root
-        with fixture:
-            self.assertTrue(root.exists())
-        self.assertTrue(fixture.destroyed)
-        self.assertFalse(root.exists())
-
-    def test_permanent_and_pre_activation_writes_are_denied(self):
-        guard = DormantOperationGuard()
-        for operation in (
-            "protected_branch.direct_push",
-            "governance.direct_push",
-            "tag.direct_write",
-            "git.force_push",
-            "production.without_owner",
-            "branch_protection.live_update",
-            "single_owner_policy.activate",
-        ):
-            with self.subTest(operation=operation), self.assertRaises(OperationDenied):
-                guard.authorize(operation)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/single_owner_governance/test_evidence_status_service.py

@@ -1,113 +0,0 @@
-from __future__ import annotations
-
-import contextlib
-import io
-import json
-import os
-from pathlib import Path
-import subprocess
-import sys
-import tempfile
-import unittest
-from unittest.mock import patch
-
-from aios_single_owner_governance.evidence import (
-    ActivationEvidenceClosureBuilder,
-    ApprovalMirrorStatusVerifier,
-)
-from aios_single_owner_governance.service import main, runtime_state
-from aios_single_owner_governance.status import (
-    DormantStatusProducer,
-    MachineStatusProducer,
-)
-
-from .helpers import HEAD
-
-
-class EvidenceStatusServiceTests(unittest.TestCase):
-    def test_complete_evidence_set_builds_canonical_closure(self):
-        evidence = {
-            "activation_prerequisites_commit_oid": "a" * 40,
-            "activation_prerequisites_tree_oid": "b" * 40,
-            "artifact_sha256": "c" * 64,
-            "rollback_artifact_sha256": "d" * 64,
-            "validator_sha256": "e" * 64,
-            "state_machine_sha256": "f" * 64,
-            "branch_protection_adapter_sha256": "0" * 64,
-            "rollback_configuration_sha256": "1" * 64,
-            "test_evidence_sha256": "2" * 64,
-            "deployment_manifest_sha256": "3" * 64,
-            "dormant_unit_manifest_sha256": "4" * 64,
-            "dormant_deployment_evidence_sha256": "5" * 64,
-        }
-        closure = ActivationEvidenceClosureBuilder().build(evidence)
-        self.assertEqual(len(closure.closure_sha256), 64)
-        with self.assertRaises(ValueError):
-            ActivationEvidenceClosureBuilder().build({})
-
-    def test_mirror_verifier_requires_all_four_refs_and_fsck(self):
-        verifier = ApprovalMirrorStatusVerifier()
-        self.assertTrue(verifier.verify(
-            HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
-            authoritative_oid=HEAD, controller_visible_oid=HEAD,
-            fsck_connectivity="PASS",
-        ))
-        self.assertFalse(verifier.verify(
-            HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
-            authoritative_oid="b" * 40, controller_visible_oid=HEAD,
-            fsck_connectivity="PASS",
-        ))
-
-    def test_machine_status_is_fixture_only_and_dormant(self):
-        with self.assertRaises(DormantStatusProducer):
-            MachineStatusProducer().produce("aios/machine-gates", "success", HEAD)
-        with self.assertRaises(DormantStatusProducer):
-            MachineStatusProducer(enabled=True, fixture=True).produce(
-                "aios/machine-gates", "success", HEAD
-            )
-        record = MachineStatusProducer(fixture=True).produce(
-            "aios/evidence-closure", "success", HEAD
-        )
-        self.assertEqual(record.head_oid, HEAD)
-
-    def test_service_self_check_is_dormant_and_activate_is_rejected(self):
-        output = io.StringIO()
-        with contextlib.redirect_stdout(output):
-            self.assertEqual(main(["--component", "owner-review-adapter", "--self-check"]), 0)
-        state = json.loads(output.getvalue())
-        self.assertFalse(state["active"])
-        self.assertFalse(state["production_mutation"])
-        with contextlib.redirect_stderr(io.StringIO()):
-            with self.assertRaises(SystemExit):
-                main(["--component", "owner-review-adapter", "--activate"])
-
-    def test_write_flags_cannot_be_enabled(self):
-        for name in (
-            "AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED",
-            "AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE",
-            "AIOS_GITOPS_WRITES_ENABLED",
-        ):
-            with self.subTest(name=name), patch.dict(os.environ, {name: "true"}, clear=False):
-                with contextlib.redirect_stderr(io.StringIO()):
-                    with self.assertRaises(SystemExit):
-                        main(["--component", "machine-status-producer", "--self-check"])
-
-    def test_service_entrypoint_restart_recovery_across_processes(self):
-        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
-            command = [
-                sys.executable, "-m", "aios_single_owner_governance.service",
-                "--component", "owner-review-adapter", "--self-check",
-                "--checkpoint-dir", root,
-            ]
-            first = subprocess.run(command, text=True, capture_output=True, check=True)
-            second = subprocess.run(command, text=True, capture_output=True, check=True)
-            first_state = json.loads(first.stdout)
-            second_state = json.loads(second.stdout)
-            self.assertFalse(first_state["restart_recovery"])
-            self.assertTrue(second_state["restart_recovery"])
-            self.assertEqual(second_state["prior_invocations"], 1)
-            self.assertFalse(second_state["production_mutation"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/single_owner_governance/test_live_activation.py

@@ -1,476 +0,0 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import sqlite3
-import subprocess
-import hashlib
-import os
-from datetime import datetime, timedelta, timezone
-
-import pytest
-
-from aios_single_owner_governance.activation import (
-    ActivationEngine,
-    InjectedCrash,
-    validate_bindings,
-)
-import aios_single_owner_governance.validator as validator_module
-import aios_single_owner_governance.state_machine as state_machine_module
-from aios_single_owner_governance.gitea_protected_branch import (
-    CASConflict,
-    FileProtectedBranchAdapter,
-    SQLiteProtectedBranchAdapter,
-    UnsafePolicy,
-    calculate_snapshot_sha256,
-    validate_target,
-)
-from aios_single_owner_governance.journal import sha256_json
-
-
-MACHINE_CONTEXTS = {
-    "aios/validator",
-    "aios/state-machine",
-    "aios/branch-protection",
-    "aios/secret-scan",
-    "aios/evidence-closure",
-}
-
-
-def write_json(path: Path, value: object) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(value, sort_keys=True) + "\n", encoding="utf-8")
-
-
-def legacy_row() -> dict:
-    return {
-        "id": 7,
-        "repo_id": 1,
-        "branch_name": "governance/approvals",
-        "can_push": 0,
-        "enable_whitelist": 0,
-        "whitelist_user_i_ds": "null",
-        "whitelist_team_i_ds": "null",
-        "enable_merge_whitelist": 1,
-        "whitelist_deploy_keys": 0,
-        "merge_whitelist_user_i_ds": "[42,43]",
-        "merge_whitelist_team_i_ds": "null",
-        "enable_status_check": 0,
-        "status_check_contexts": "[]",
-        "enable_approvals_whitelist": 1,
-        "approvals_whitelist_user_i_ds": "[42,43]",
-        "approvals_whitelist_team_i_ds": "null",
-        "required_approvals": 2,
-        "block_on_rejected_reviews": 1,
-        "block_on_official_review_requests": 1,
-        "block_on_outdated_branch": 1,
-        "dismiss_stale_approvals": 1,
-        "ignore_stale_approvals": 0,
-        "require_signed_commits": 0,
-        "protected_file_patterns": "",
-        "unprotected_file_patterns": "",
-        "created_unix": 1,
-        "updated_unix": 10,
-    }
-
-
-def owner_row() -> dict:
-    row = legacy_row()
-    row.update(
-        {
-            "approvals_whitelist_user_i_ds": "[2]",
-            "required_approvals": 1,
-            "enable_status_check": 1,
-            "status_check_contexts": json.dumps(sorted(MACHINE_CONTEXTS), separators=(",", ":")),
-            "updated_unix": 11,
-        }
-    )
-    return row
-
-
-def git_output(git_dir: Path, *args: str, stdin: str | None = None) -> str:
-    return subprocess.run(
-        ["git", f"--git-dir={git_dir}", *args], input=stdin, text=True,
-        capture_output=True, check=True,
-        env={
-            **__import__("os").environ,
-            "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.invalid",
-            "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.invalid",
-        },
-    ).stdout.strip()
-
-
-def evidence_fixture(tmp_path: Path, execution_binding: dict) -> dict:
-    git_dir = tmp_path / "evidence.git"
-    subprocess.run(["git", "init", "--bare", str(git_dir)], capture_output=True, check=True)
-    closure_oid = git_output(git_dir, "hash-object", "-w", "--stdin", stdin="closure\n")
-    head_oid = git_output(git_dir, "hash-object", "-w", "--stdin", stdin="head\n")
-    expiry = datetime.now(timezone.utc) + timedelta(hours=1)
-    execution_binding = {**execution_binding, "evidence_closure_oid": closure_oid}
-    manifest = {
-        "approval_type": "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION",
-        "expires_at": expiry.isoformat().replace("+00:00", "Z"),
-        "execution_binding": execution_binding,
-        "execution_binding_sha256": sha256_json(execution_binding),
-    }
-    manifest_raw = json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n"
-    manifest_blob = git_output(git_dir, "hash-object", "-w", "--stdin", stdin=manifest_raw)
-    leaf = git_output(git_dir, "mktree", stdin=f"100644 blob {manifest_blob}\tactivation.json\n")
-    approvals = git_output(git_dir, "mktree", stdin=f"040000 tree {leaf}\tsingle-owner\n")
-    tree = git_output(git_dir, "mktree", stdin=f"040000 tree {approvals}\tapprovals\n")
-    merge_oid = git_output(git_dir, "commit-tree", tree, stdin="approved\n")
-    git_output(git_dir, "update-ref", "refs/heads/governance/approvals", merge_oid)
-
-    db = tmp_path / "approval.db"
-    with sqlite3.connect(db) as connection:
-        connection.executescript(
-            "CREATE TABLE issue(id INTEGER,repo_id INTEGER,is_closed INTEGER);"
-            "CREATE TABLE pull_request(issue_id INTEGER,\"index\" INTEGER,base_branch TEXT,has_merged INTEGER,merged_commit_id TEXT);"
-            "CREATE TABLE user(id INTEGER,lower_name TEXT);"
-            "CREATE TABLE review(id INTEGER,type INTEGER,reviewer_id INTEGER,issue_id INTEGER,commit_id TEXT,stale INTEGER,dismissed INTEGER,created_unix INTEGER);"
-            "CREATE TABLE commit_status(id INTEGER,repo_id INTEGER,state TEXT,sha TEXT,context TEXT);"
-        )
-        connection.execute("INSERT INTO issue VALUES(10,1,1)")
-        connection.execute("INSERT INTO pull_request VALUES(10,31,'governance/approvals',1,?)", (merge_oid,))
-        connection.execute("INSERT INTO user VALUES(2,'owner')")
-        connection.execute(
-            "INSERT INTO review VALUES(52,1,2,10,?,0,0,?)",
-            (head_oid, int(datetime.now(timezone.utc).timestamp())),
-        )
-        for index, context in enumerate(sorted(MACHINE_CONTEXTS), 1):
-            connection.execute("INSERT INTO commit_status VALUES(?,1,'success',?,?)", (index, head_oid, context))
-    return {
-        "git_dir": git_dir, "merge_oid": merge_oid, "closure_oid": closure_oid,
-        "head_oid": head_oid, "db": db, "expiry": expiry.isoformat().replace("+00:00", "Z"),
-        "manifest_path": "approvals/single-owner/activation.json",
-        "manifest_blob": manifest_blob,
-        "manifest_sha256": hashlib.sha256(manifest_raw.encode()).hexdigest(),
-        "execution_binding": execution_binding,
-    }
-
-
-def bindings(tmp_path: Path, adapter: FileProtectedBranchAdapter, **overrides) -> dict:
-    validator_path = Path(validator_module.__file__).resolve()
-    state_machine_path = Path(state_machine_module.__file__).resolve()
-    validator_sha = hashlib.sha256(validator_path.read_bytes()).hexdigest()
-    state_machine_sha = hashlib.sha256(state_machine_path.read_bytes()).hexdigest()
-    execution_binding = {
-        "task_id": "AIOS-LIVE-ACTIVATION-TEST",
-        "operation_id": "op-live-001",
-        "expected_validator_sha256": validator_sha,
-        "expected_state_machine_sha256": state_machine_sha,
-        "expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
-        "expected_gitea_schema_sha256": "0" * 64,
-        "target_branch_protection_sha256": calculate_snapshot_sha256(owner_row()),
-        "backend": "file",
-        "repo_id": 1,
-        "branch": "governance/approvals",
-        "requester": "agent",
-        "executor": "activation-executor",
-        "forbidden_operations": {
-            "main": False, "governance_direct": False, "integration": False,
-            "tags": False, "force_push": False,
-        },
-    }
-    evidence = evidence_fixture(tmp_path, execution_binding)
-    value = {
-        "task_id": "AIOS-LIVE-ACTIVATION-TEST",
-        "operation_id": "op-live-001",
-        "expected_governance_oid": evidence["merge_oid"],
-        "observed_governance_oid": evidence["merge_oid"],
-        "expected_validator_sha256": validator_sha,
-        "observed_validator_sha256": validator_sha,
-        "expected_state_machine_sha256": state_machine_sha,
-        "observed_state_machine_sha256": state_machine_sha,
-        "validator_path": str(validator_path),
-        "state_machine_path": str(state_machine_path),
-        "governance_git_dir": str(evidence["git_dir"].resolve()),
-        "governance_ref": "refs/heads/governance/approvals",
-        "evidence_git_dir": str(evidence["git_dir"].resolve()),
-        "approval_db_path": str(evidence["db"].resolve()),
-        "expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
-        "activation_approval_merge_oid": evidence["merge_oid"],
-        "approval_request_head_oid": evidence["head_oid"],
-        "activation_approval_pr_number": 31,
-        "activation_manifest_path": evidence["manifest_path"],
-        "activation_manifest_blob_oid": evidence["manifest_blob"],
-        "activation_manifest_sha256": evidence["manifest_sha256"],
-        "expires_at": evidence["expiry"],
-        "evidence_closure_oid": evidence["closure_oid"],
-        "observed_evidence_closure_oid": evidence["closure_oid"],
-        "owner_review_oid": 52,
-        "observed_owner_review_oid": 52,
-        "requester": "agent",
-        "executor": "activation-executor",
-        "wip": False,
-        "machine_checks": {
-            context: {"state": "success", "head_oid": evidence["head_oid"]}
-            for context in MACHINE_CONTEXTS
-        },
-        "forbidden_operations": {
-            "main": False,
-            "governance_direct": False,
-            "integration": False,
-            "tags": False,
-            "force_push": False,
-        },
-        "target_branch_protection": owner_row(),
-        "backend": "file",
-        "backend_path": str(adapter.path.resolve()),
-        "repo_id": 1,
-        "branch": "governance/approvals",
-        "expected_gitea_schema_sha256": "0" * 64,
-    }
-    value.update(overrides)
-    return value
-
-
-def harness(tmp_path: Path):
-    backend = tmp_path / "protected.json"
-    write_json(backend, legacy_row())
-    adapter = FileProtectedBranchAdapter(backend)
-    engine = ActivationEngine(adapter, tmp_path / "state", tmp_path / "run" / "activation.lock")
-    request = bindings(tmp_path, adapter)
-    return backend, adapter, engine, request
-
-
-def test_live_plan_success(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    result = engine.plan(request)
-    assert result["status"] == "PLAN_READY"
-    assert len(result["plan_id"]) == 64
-
-
-def test_live_activate_verify_status_success(tmp_path: Path) -> None:
-    _, adapter, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    activated = engine.activate(request, planned["plan_id"])
-    assert activated["status"] == "ACTIVATED"
-    assert adapter.read_current() == owner_row()
-    assert engine.verify(request, request["operation_id"])["status"] == "VERIFIED"
-    assert engine.status(request["operation_id"])["transaction"]["phase"] == "COMMITTED"
-    assert (engine.journal.root / "current-snapshot.json").exists()
-    assert (engine.journal.root / "rollback-snapshot.json").exists()
-
-
-def test_live_rollback_success(tmp_path: Path) -> None:
-    _, adapter, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    engine.activate(request, planned["plan_id"])
-    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ROLLED_BACK"
-    assert adapter.read_current() == legacy_row()
-    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
-
-
-def test_missing_approval_denied(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request.pop("owner_review_oid")
-    with pytest.raises(RuntimeError, match="MISSING_BINDINGS"):
-        engine.plan(request)
-
-
-@pytest.mark.parametrize(
-    ("field", "value", "message"),
-    [
-        ("observed_owner_review_oid", 99, "WRONG_APPROVAL_OID_DENIED"),
-        ("observed_governance_oid", "f" * 40, "WRONG_GOVERNANCE_OID_DENIED"),
-        ("observed_evidence_closure_oid", "f" * 40, "WRONG_EVIDENCE_CLOSURE_DENIED"),
-    ],
-)
-def test_wrong_binding_denied(tmp_path: Path, field: str, value: object, message: str) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request[field] = value
-    with pytest.raises(RuntimeError, match=message):
-        engine.plan(request)
-
-
-def test_cas_conflict_denied(tmp_path: Path) -> None:
-    backend, _, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    changed = legacy_row()
-    changed["updated_unix"] = 12
-    write_json(backend, changed)
-    with pytest.raises(CASConflict):
-        engine.activate(request, planned["plan_id"])
-
-
-@pytest.mark.parametrize("crash_at", ["before_apply", "during_apply", "after_apply_before_receipt", "after_commit_wal"])
-def test_process_crash_restart_recovery(tmp_path: Path, crash_at: str) -> None:
-    _, adapter, engine, request = harness(tmp_path)
-    request["crash_at"] = crash_at
-    planned = engine.plan(request)
-    with pytest.raises(InjectedCrash):
-        engine.activate(request, planned["plan_id"])
-    request.pop("crash_at")
-    result = engine.activate(request, planned["plan_id"])
-    expected_status = "ALREADY_COMMITTED" if crash_at == "after_commit_wal" else "ACTIVATED"
-    assert result["status"] == expected_status
-    assert adapter.read_current() == owner_row()
-    assert engine.journal.receipt_path(request["operation_id"]).exists()
-    assert engine.journal.audit.has_event(request["operation_id"], "SINGLE_OWNER_ACTIVATION_COMMITTED")
-
-
-def test_post_activation_verify_failure_rolls_back(tmp_path: Path, monkeypatch) -> None:
-    _, adapter, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    monkeypatch.setattr(adapter, "verify_applied", lambda _: False)
-    with pytest.raises(RuntimeError, match="POST_ACTIVATION_VERIFY_FAILED"):
-        engine.activate(request, planned["plan_id"])
-    assert adapter.read_current() == legacy_row()
-    assert engine.journal.load(request["operation_id"])["phase"] == "ROLLED_BACK"
-
-
-def test_operation_id_idempotency(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    engine.activate(request, planned["plan_id"])
-    assert engine.activate(request, planned["plan_id"])["status"] == "ALREADY_COMMITTED"
-
-
-@pytest.mark.parametrize("crash_at", ["after_rollback_cas", "after_rollback_wal"])
-def test_rollback_crash_restart_repairs_closure(tmp_path: Path, crash_at: str) -> None:
-    _, adapter, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    engine.activate(request, planned["plan_id"])
-    request["crash_at"] = crash_at
-    with pytest.raises(InjectedCrash):
-        engine.rollback(request, request["operation_id"], planned["plan_id"])
-    request.pop("crash_at")
-    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
-    assert adapter.read_current() == legacy_row()
-    assert engine.journal.receipt_path(request["operation_id"]).exists()
-    assert engine.journal.audit.has_event(request["operation_id"], "ACTIVATION_ROLLED_BACK")
-
-
-def test_operation_id_payload_mismatch_denied(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    planned = engine.plan(request)
-    engine.activate(request, planned["plan_id"])
-    changed = dict(request)
-    approval_alias = tmp_path / "approval-alias.db"
-    os.link(request["approval_db_path"], approval_alias)
-    changed["approval_db_path"] = str(approval_alias)
-    with pytest.raises(RuntimeError, match="PAYLOAD_MISMATCH"):
-        engine.activate(changed, planned["plan_id"])
-
-
-@pytest.mark.parametrize(
-    ("field", "value"),
-    [("operation_id", "op-live-REPLAY"), ("task_id", "REPLAY-TASK")],
-)
-def test_approved_execution_binding_replay_denied(tmp_path: Path, field: str, value: str) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request[field] = value
-    with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
-        engine.plan(request)
-
-
-def test_expiry_must_come_from_approved_manifest(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request["expires_at"] = (datetime.now(timezone.utc) + timedelta(hours=12)).isoformat().replace("+00:00", "Z")
-    with pytest.raises(RuntimeError, match="APPROVAL_EXPIRY_BINDING_MISMATCH"):
-        engine.plan(request)
-
-
-def test_target_snapshot_must_match_approved_manifest(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request["target_branch_protection"] = {**owner_row(), "updated_unix": 99}
-    with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
-        engine.plan(request)
-
-
-def test_approval_database_must_be_backend_database(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    other_db = tmp_path / "other.db"
-    other_db.write_bytes(Path(request["approval_db_path"]).read_bytes())
-    request.update(backend="sqlite", backend_path=str(other_db.resolve()))
-    with pytest.raises(RuntimeError, match="APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH"):
-        engine.plan(request)
-
-
-def test_audit_chain_continuity_and_credential_redaction(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    request["api_token"] = "must-not-appear"
-    planned = engine.plan(request)
-    engine.activate(request, planned["plan_id"])
-    assert len(engine.journal.audit.verify()) == 64
-    raw = (engine.journal.root / "audit.jsonl").read_text(encoding="utf-8")
-    assert "must-not-appear" not in raw
-    assert "[REDACTED]" not in raw  # only hashes/events are persisted to audit
-
-
-@pytest.mark.parametrize(
-    "mutation",
-    [
-        lambda request: request.update(requester="owner"),
-        lambda request: request.update(executor="owner"),
-        lambda request: request.update(wip=True),
-        lambda request: request.update(machine_checks={"anything": {"state": "success", "head_oid": request["approval_request_head_oid"]}}),
-        lambda request: request.update(forbidden_operations={"main": True}),
-        lambda request: request.update(forbidden_operations={"tags": True}),
-        lambda request: request.update(forbidden_operations={"force_push": True}),
-    ],
-)
-def test_negative_governance_acceptance(tmp_path: Path, mutation) -> None:
-    _, _, engine, request = harness(tmp_path)
-    mutation(request)
-    with pytest.raises(RuntimeError):
-        engine.plan(request)
-
-
-@pytest.mark.parametrize(
-    "mutation",
-    [
-        lambda target: target.update(enable_approvals_whitelist=0),
-        lambda target: target.update(enable_merge_whitelist=0),
-        lambda target: target.update(status_check_contexts="[]"),
-    ],
-)
-def test_unsafe_policy_enable_fields_denied(tmp_path: Path, mutation) -> None:
-    _, _, _, request = harness(tmp_path)
-    mutation(request["target_branch_protection"])
-    with pytest.raises(UnsafePolicy):
-        validate_target(legacy_row(), request["target_branch_protection"])
-
-
-@pytest.mark.parametrize("invalid_owner", ["codex", "gpt"])
-def test_owner_review_must_come_from_gitea_database(tmp_path: Path, invalid_owner: str) -> None:
-    _, _, engine, request = harness(tmp_path)
-    with sqlite3.connect(request["approval_db_path"]) as connection:
-        connection.execute("UPDATE user SET lower_name=? WHERE id=2", (invalid_owner,))
-    with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
-        engine.plan(request)
-
-
-def test_stale_owner_review_denied_from_gitea_database(tmp_path: Path) -> None:
-    _, _, engine, request = harness(tmp_path)
-    with sqlite3.connect(request["approval_db_path"]) as connection:
-        connection.execute("UPDATE review SET stale=1 WHERE id=52")
-    with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
-        engine.plan(request)
-
-
-def create_sqlite_fixture(path: Path) -> None:
-    row = legacy_row()
-    columns = []
-    for key, value in row.items():
-        kind = "INTEGER" if isinstance(value, int) else "TEXT"
-        columns.append(f"{key} {kind}")
-    with sqlite3.connect(path) as connection:
-        connection.execute(f"CREATE TABLE protected_branch ({','.join(columns)})")
-        connection.execute(
-            f"INSERT INTO protected_branch ({','.join(row)}) VALUES ({','.join('?' for _ in row)})",
-            list(row.values()),
-        )
-
-
-def test_sqlite_full_row_cas_and_rollback(tmp_path: Path) -> None:
-    db = tmp_path / "gitea.db"
-    create_sqlite_fixture(db)
-    probe = SQLiteProtectedBranchAdapter(db, expected_schema_sha256="0" * 64)
-    adapter = SQLiteProtectedBranchAdapter(db, expected_schema_sha256=probe.schema_sha256())
-    previous = adapter.read_current()
-    applied = adapter.apply_atomic(calculate_snapshot_sha256(previous), owner_row())
-    assert applied == owner_row()
-    restored = adapter.rollback_atomic(calculate_snapshot_sha256(applied), previous)
-    assert restored == previous

tests/single_owner_governance/test_state_cas_rollback.py

@@ -1,126 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-import tempfile
-import unittest
-
-from aios_single_owner_governance.fixture import DisposableGiteaFixture
-from aios_single_owner_governance.lock import MaintenanceLock
-from aios_single_owner_governance.models import GovernanceState
-from aios_single_owner_governance.protection import (
-    BranchProtectionCASMigrationAdapter,
-    CASConflict,
-    MemoryProtectionStore,
-    snapshot_sha256,
-)
-from aios_single_owner_governance.rollback import (
-    ProtectionCompensationSaga,
-    RollbackConfigurationManager,
-)
-from aios_single_owner_governance.state_machine import GovernanceStateMachine, InvalidTransition
-
-from .helpers import legacy_protection, single_owner_protection
-from .helpers import context
-from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
-
-
-class StateCasRollbackTests(unittest.TestCase):
-    def test_ordinary_owner_approved_merge_simulation(self):
-        result = SingleOwnerApprovalValidator().validate(context())
-        self.assertEqual(result.owner_review.reviewer, "owner")
-        machine = GovernanceStateMachine()
-        for target in (
-            GovernanceState.OWNER_APPROVAL_VERIFIED,
-            GovernanceState.MACHINE_GATES_VERIFIED,
-            GovernanceState.AUTO_MERGE_READY,
-            GovernanceState.OWNER_APPROVED_EXECUTION_READY,
-        ):
-            machine.transition(target)
-        self.assertEqual(machine.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
-
-    def test_validator_failure_is_stateless_and_recoverable(self):
-        validator = SingleOwnerApprovalValidator()
-        with self.assertRaises(ValidationDenied):
-            validator.validate(context(wip=True))
-        self.assertEqual(validator.validate(context()).owner_review.reviewer, "owner")
-
-    def test_state_machine_happy_path_and_restart_recovery(self):
-        machine = GovernanceStateMachine()
-        for target in (
-            GovernanceState.OWNER_APPROVAL_VERIFIED,
-            GovernanceState.MACHINE_GATES_VERIFIED,
-            GovernanceState.AUTO_MERGE_READY,
-            GovernanceState.OWNER_APPROVED_EXECUTION_READY,
-        ):
-            machine.transition(target)
-        restarted = GovernanceStateMachine(machine.state)
-        self.assertEqual(restarted.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
-
-    def test_state_machine_rollback_to_checkpoint(self):
-        machine = GovernanceStateMachine()
-        machine.transition(GovernanceState.OWNER_APPROVAL_VERIFIED)
-        checkpoint = machine.state
-        machine.transition(GovernanceState.MACHINE_GATES_VERIFIED)
-        machine.restore(checkpoint)
-        self.assertEqual(machine.state, GovernanceState.OWNER_APPROVAL_VERIFIED)
-
-    def test_service_restart_from_each_checkpoint(self):
-        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
-            checkpoint = Path(root) / "state.json"
-            for state in GovernanceState:
-                with self.subTest(state=state):
-                    GovernanceStateMachine(state).save_fixture(checkpoint)
-                    restarted = GovernanceStateMachine.load_fixture(checkpoint)
-                    self.assertEqual(restarted.state, state)
-
-    def test_invalid_transition_denied(self):
-        with self.assertRaises(InvalidTransition):
-            GovernanceStateMachine().transition(GovernanceState.AUTO_MERGE_READY)
-
-    def test_branch_protection_cas_success_and_rollback(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old, new = legacy_protection(), single_owner_protection()
-        plan = adapter.plan(old, new, snapshot_sha256(old))
-        store = MemoryProtectionStore(old)
-        self.assertEqual(adapter.apply_fixture(store, plan), new)
-        self.assertEqual(adapter.rollback_fixture(store, plan), old)
-
-    def test_branch_protection_cas_conflict_fails_closed(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old, changed, new = legacy_protection(), legacy_protection(7), single_owner_protection()
-        plan = adapter.plan(old, new, snapshot_sha256(old))
-        store = MemoryProtectionStore(changed)
-        with self.assertRaises(CASConflict):
-            adapter.apply_fixture(store, plan)
-
-    def test_partial_activation_recovery_saga(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old, new = legacy_protection(), single_owner_protection()
-        plan = adapter.plan(old, new, snapshot_sha256(old))
-        saga = ProtectionCompensationSaga(adapter, MemoryProtectionStore(old), plan)
-        saga.execute_fixture()
-        self.assertEqual(saga.compensate_fixture(), old)
-        self.assertTrue(saga.applied and saga.rolled_back)
-
-    def test_rollback_configuration_fixture_install(self):
-        adapter = BranchProtectionCASMigrationAdapter()
-        old, new = legacy_protection(), single_owner_protection()
-        plan = adapter.plan(old, new, snapshot_sha256(old))
-        manager = RollbackConfigurationManager()
-        config = manager.build(plan)
-        with DisposableGiteaFixture(old) as fixture:
-            path = manager.install_fixture(fixture.root, config)
-            self.assertTrue(path.exists())
-            self.assertTrue(manager.verify(config, plan))
-
-    def test_maintenance_lock_fences_concurrent_holder(self):
-        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
-            path = Path(root) / "maintenance.lock"
-            with MaintenanceLock(path):
-                with self.assertRaises(BlockingIOError):
-                    with MaintenanceLock(path):
-                        pass
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/single_owner_governance/test_systemd_contract.py

@@ -1,45 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-import unittest
-
-
-ROOT = Path(__file__).resolve().parents[2]
-
-
-class SystemdContractTests(unittest.TestCase):
-    def test_units_are_static_dormant_and_credential_free(self):
-        units = sorted(
-            unit for unit in (ROOT / "deploy" / "systemd").glob("aios-governance-*.service")
-            if unit.name != "aios-governance-single-owner-activation.service"
-        )
-        self.assertEqual(len(units), 2)
-        for unit in units:
-            text = unit.read_text(encoding="utf-8")
-            with self.subTest(unit=unit.name):
-                self.assertIn("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false", text)
-                self.assertIn("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false", text)
-                self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
-                self.assertIn("ProtectSystem=strict", text)
-                self.assertNotIn("EnvironmentFile=", text)
-                self.assertNotIn("LoadCredential=", text)
-                self.assertNotIn("WantedBy=", text)
-                self.assertNotIn("[Install]", text)
-                self.assertIn("--checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance", text)
-                self.assertNotIn("/data/aios-gitops/production", text)
-
-    def test_activation_verifier_is_static_and_has_no_credential_or_install_hook(self):
-        unit = ROOT / "deploy" / "systemd" / "aios-governance-single-owner-activation.service"
-        text = unit.read_text(encoding="utf-8")
-        self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
-        self.assertIn("--verify", text)
-        self.assertIn("ProtectSystem=strict", text)
-        self.assertNotIn("EnvironmentFile=", text)
-        self.assertNotIn("LoadCredential=", text)
-        self.assertNotIn("WantedBy=", text)
-        self.assertNotIn("[Install]", text)
-        self.assertNotIn("--activate", text)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/single_owner_governance/test_validator.py

@@ -1,77 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import replace
-from datetime import timedelta
-import unittest
-
-from aios_single_owner_governance.models import Review
-from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
-
-from .helpers import HEAD, NOW, context, gates, owner_review
-
-
-class ValidatorTests(unittest.TestCase):
-    def setUp(self):
-        self.validator = SingleOwnerApprovalValidator()
-
-    def denied(self, code, **changes):
-        with self.assertRaises(ValidationDenied) as caught:
-            self.validator.validate(context(**changes))
-        self.assertEqual(caught.exception.code, code)
-
-    def test_owner_current_head_and_machine_gates_pass(self):
-        result = self.validator.validate(context())
-        self.assertEqual(result.owner_review.reviewer, "owner")
-        self.assertEqual(len(result.machine_contexts), 3)
-
-    def test_requester_self_approval_denied(self):
-        self.denied("REQUESTER_SELF_APPROVAL", requester="owner")
-
-    def test_executor_self_approval_denied(self):
-        self.denied("EXECUTOR_SELF_APPROVAL", executor="owner")
-
-    def test_codex_as_owner_denied(self):
-        codex = owner_review(reviewer="codex", reviewer_id=42)
-        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(codex,))
-
-    def test_gpt_as_owner_denied(self):
-        gpt = owner_review(reviewer="gpt", reviewer_id=43)
-        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(gpt,))
-
-    def test_non_owner_approval_denied(self):
-        other = owner_review(reviewer="other", reviewer_id=77)
-        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(other,))
-
-    def test_stale_owner_approval_denied(self):
-        self.denied("STALE_OWNER_APPROVAL", reviews=(owner_review(stale=True),))
-
-    def test_dismissed_owner_approval_denied(self):
-        self.denied("DISMISSED_OWNER_APPROVAL", reviews=(owner_review(dismissed=True),))
-
-    def test_old_head_owner_approval_denied(self):
-        self.denied("OWNER_APPROVAL_NOT_LATEST_HEAD", reviews=(owner_review(head_oid="b" * 40),))
-
-    def test_wip_denied(self):
-        self.denied("WIP_BLOCKED", wip=True)
-
-    def test_machine_gate_failure_denied(self):
-        self.denied("MACHINE_GATE_FAILED", machine_gates=gates(state="failure"))
-
-    def test_secret_scan_failure_denied(self):
-        failed = list(gates())
-        failed[1] = replace(failed[1], state="failure")
-        self.denied("MACHINE_GATE_FAILED", machine_gates=tuple(failed))
-
-    def test_machine_gate_wrong_head_denied(self):
-        self.denied("MACHINE_GATE_NOT_LATEST_HEAD", machine_gates=gates(head_oid="c" * 40))
-
-    def test_expired_request_denied(self):
-        self.denied("APPROVAL_EXPIRED", now=NOW + timedelta(hours=2))
-
-    def test_review_after_expiry_denied(self):
-        late = owner_review(reviewed_at=NOW + timedelta(hours=2))
-        self.denied("OWNER_APPROVAL_EXPIRED", reviews=(late,))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_gitea_branch_read_projection.py

@@ -0,0 +1,24 @@
+import importlib.util
+import unittest
+import sys
+from pathlib import Path
+
+
+def module():
+    p = Path(__file__).parents[1] / "active-runtime/app/aios_isolated_writer_api/integration_refresh.py"
+    s = importlib.util.spec_from_file_location("projection_under_test", p); m = importlib.util.module_from_spec(s); sys.modules[s.name] = m; s.loader.exec_module(m); return m
+
+
+class GiteaBranchReadProjectionTests(unittest.TestCase):
+    def test_auth_and_timeout_fail_closed(self):
+        m = module()
+        for reply in ({"status": 401}, {"status": 403}, {"status": 500}, {"status": 200, "oid": "short"}, {"status": 200, "oid": "z" * 40}):
+            with self.subTest(reply=reply), self.assertRaises(m.ProvisioningError) as caught:
+                m.GiteaBranchReadProjection(lambda repo, branch: reply).head("aios/aios", "integration/x")
+            self.assertEqual(503, caught.exception.status)
+        def timeout(repo, branch): raise TimeoutError()
+        with self.assertRaisesRegex(m.ProvisioningError, "gitea_projection_timeout"):
+            m.GiteaBranchReadProjection(timeout).head("aios/aios", "integration/x")
+
+
+if __name__ == "__main__": unittest.main()

tests/test_integration_refresh_provisioner.py

@@ -0,0 +1,95 @@
+import importlib.util
+import tempfile
+import sys
+import unittest
+from dataclasses import replace
+import json
+from pathlib import Path
+
+
+def load():
+    path = Path(__file__).parents[1] / "active-runtime/app/aios_isolated_writer_api/integration_refresh.py"
+    spec = importlib.util.spec_from_file_location("integration_refresh", path)
+    module = importlib.util.module_from_spec(spec); sys.modules[spec.name] = module; spec.loader.exec_module(module)
+    return module
+
+
+class RefreshProvisionerTests(unittest.TestCase):
+    def setUp(self):
+        self.m = load(); self.temp = tempfile.TemporaryDirectory(); self.root = Path(self.temp.name)
+        self.head = "a" * 40
+        self.binding = self.m.Binding("task-7", "aios/aios", "integration/refresh", "actor-1", "writer.service", ("docs/release.json",), 2)
+        self.grants = self.m.GrantStore(self.root / "grants.sqlite")
+        self.p = self.m.IntegrationRefreshProvisioner(
+            grants=self.grants,
+            projection=self.m.GiteaBranchReadProjection(lambda repo, branch: {"status": 200, "oid": self.head}),
+            ledger_path=self.root / "ledger.sqlite", audit=self.m.AppendOnlyAudit(self.root / "audit.jsonl"), approved_binding=self.binding)
+
+    def tearDown(self): self.temp.cleanup()
+
+    def test_bound_grant_idempotency_quota_rotation_and_audit(self):
+        old = self.grants.issue(self.binding); new = self.grants.rotate(old, self.binding, revocation_probe=lambda token: 401)
+        with self.assertRaisesRegex(self.m.ProvisioningError, "credential_rejected"):
+            self.grants.consume(old, self.binding, "docs/release.json")
+        def apply(): self.head = "b" * 40; return self.head
+        answer = self.p.refresh(token=new, binding=self.binding, path="docs/release.json", operation_id="refresh-1", target_head_oid="a" * 40, apply=apply)
+        self.assertEqual("COMPLETED", answer["state"])
+        again = self.p.refresh(token=new, binding=self.binding, path="docs/release.json", operation_id="refresh-1", target_head_oid="a" * 40, apply=lambda: self.fail("must not apply"))
+        self.assertTrue(again["idempotent"])
+        audit = [json.loads(line) for line in (self.root / "audit.jsonl").read_text().splitlines()]
+        self.assertEqual("actor-1", audit[-1]["actor_identity"])
+        self.assertEqual("0" * 64, audit[0]["previous_event_hash"])
+        for index, event in enumerate(audit):
+            stored = event["event_hash"]
+            unsigned = dict(event); unsigned.pop("event_hash")
+            self.assertEqual(self.m._canonical(unsigned), stored)
+            if index:
+                self.assertEqual(audit[index - 1]["event_hash"], event["previous_event_hash"])
+        with self.assertRaisesRegex(self.m.ProvisioningError, "operation_id_reuse_rejected"):
+            self.p.refresh(token=new, binding=self.binding, path="docs/release.json", operation_id="refresh-1", target_head_oid="b" * 40, apply=lambda: "c" * 40)
+        changed_bindings = (
+            replace(self.binding, task_id="wrong"), replace(self.binding, repository="wrong/repo"),
+            replace(self.binding, branch="integration/other"), replace(self.binding, actor_identity="other"),
+            replace(self.binding, audience="other.service"), replace(self.binding, paths=("docs/other.json",)),
+        )
+        for index, changed in enumerate(changed_bindings):
+            with self.assertRaisesRegex(self.m.ProvisioningError, "grant_binding_rejected"):
+                self.p.refresh(token=new, binding=changed, path="docs/release.json", operation_id=f"wrong-{index}", target_head_oid="b" * 40, apply=lambda: "b" * 40)
+
+    def test_rotation_and_oid_validation_fail_closed(self):
+        old = self.grants.issue(self.binding)
+        with self.assertRaisesRegex(self.m.ProvisioningError, "revocation_probe_rejected"):
+            self.grants.rotate(old, self.binding, revocation_probe=lambda token: 403)
+        old = self.grants.issue(self.binding)
+        def timeout(token): raise TimeoutError()
+        with self.assertRaisesRegex(self.m.ProvisioningError, "revocation_probe_timeout"):
+            self.grants.rotate(old, self.binding, revocation_probe=timeout)
+        token = self.grants.issue(self.binding)
+        with self.assertRaisesRegex(self.m.ProvisioningError, "refresh_request_rejected"):
+            self.p.refresh(token=token, binding=self.binding, path="docs/release.json", operation_id="invalid-oid", target_head_oid="z" * 40, apply=lambda: "b" * 40)
+
+    def test_stale_target_and_remote_concurrent_advance_are_closed(self):
+        token = self.grants.issue(self.binding)
+        with self.assertRaisesRegex(self.m.ProvisioningError, "stale_target_head"):
+            self.p.refresh(token=token, binding=self.binding, path="docs/release.json", operation_id="refresh-2", target_head_oid="c" * 40, apply=lambda: "d" * 40)
+        token = self.grants.issue(self.binding)
+        calls = ["a" * 40, "f" * 40]
+        self.p.projection = self.m.GiteaBranchReadProjection(lambda repo, branch: {"status": 200, "oid": calls.pop(0)})
+        with self.assertRaisesRegex(self.m.ProvisioningError, "remote_concurrent_advance"):
+            self.p.refresh(token=token, binding=self.binding, path="docs/release.json", operation_id="refresh-3", target_head_oid="a" * 40, apply=lambda: "e" * 40)
+
+    def test_recoverable_crash_retries_without_spending_a_second_quota(self):
+        binding = self.m.Binding("task-7", "aios/aios", "integration/refresh", "actor-1", "writer.service", ("docs/release.json",), 1)
+        self.p.approved_binding = binding
+        token = self.grants.issue(binding)
+        attempts = [RuntimeError("crash"), "b" * 40]
+        def apply():
+            next_value = attempts.pop(0)
+            if isinstance(next_value, Exception): raise next_value
+            self.head = next_value; return next_value
+        with self.assertRaisesRegex(self.m.ProvisioningError, "refresh_apply_failed"):
+            self.p.refresh(token=token, binding=binding, path="docs/release.json", operation_id="refresh-4", target_head_oid="a" * 40, apply=apply)
+        self.assertEqual("COMPLETED", self.p.refresh(token=token, binding=binding, path="docs/release.json", operation_id="refresh-4", target_head_oid="a" * 40, apply=apply)["state"])
+
+
+if __name__ == "__main__": unittest.main()

tests/test_isolated_writer_integration_merge.py

@@ -0,0 +1,75 @@
+import importlib.util
+import subprocess
+import tempfile
+import unittest
+import sys
+from pathlib import Path
+
+
+def run(cwd, *args): return subprocess.check_output(["git", *args], cwd=cwd, text=True).strip()
+def mod():
+    p = Path(__file__).parents[1] / "active-runtime/app/aios_isolated_writer_api/integration_merge_writer.py"
+    s = importlib.util.spec_from_file_location("merge_under_test", p); m = importlib.util.module_from_spec(s); sys.modules[s.name] = m; s.loader.exec_module(m); return m
+
+
+class IntegrationMergeWriterTests(unittest.TestCase):
+    def setUp(self):
+        self.m = mod(); self.temp = tempfile.TemporaryDirectory(); self.root = Path(self.temp.name); self.remote = self.root / "remote.git"; self.work = self.root / "work"
+        subprocess.run(["git", "init", "--bare", self.remote], check=True, capture_output=True); subprocess.run(["git", "init", self.work], check=True, capture_output=True)
+        for k, v in (("user.email", "t@example.invalid"), ("user.name", "T")): subprocess.run(["git", "config", k, v], cwd=self.work, check=True)
+        (self.work / "a").write_text("base"); subprocess.run(["git", "add", "."], cwd=self.work, check=True); subprocess.run(["git", "commit", "-m", "base"], cwd=self.work, check=True, capture_output=True)
+        subprocess.run(["git", "branch", "-M", "integration/x"], cwd=self.work, check=True); subprocess.run(["git", "remote", "add", "origin", self.remote], cwd=self.work, check=True); subprocess.run(["git", "push", "origin", "integration/x"], cwd=self.work, check=True, capture_output=True)
+        subprocess.run(["git", "checkout", "-b", "feature/x"], cwd=self.work, check=True, capture_output=True); (self.work / "b").write_text("source"); subprocess.run(["git", "add", "."], cwd=self.work, check=True); subprocess.run(["git", "commit", "-m", "source"], cwd=self.work, check=True, capture_output=True); self.source = run(self.work, "rev-parse", "HEAD"); subprocess.run(["git", "checkout", "integration/x"], cwd=self.work, check=True, capture_output=True)
+
+    def tearDown(self): self.temp.cleanup()
+    def test_no_ff_parent_order_and_scope_rejection(self):
+        remote_head = lambda branch: run(self.work, "ls-remote", "--heads", "origin", "refs/heads/" + branch).split()[0]
+        writer = self.m.IntegrationMergeWriter(self.work, "origin", remote_head)
+        result = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
+        self.assertEqual("COMPLETED", result["state"])
+        self.assertEqual([result["pre_merge_oid"], self.source], run(self.work, "show", "-s", "--format=%P", "HEAD").split())
+        for integration, source, oid in (
+            ("main", "feature/x", self.source),
+            ("integration/x", "governance/approvals", self.source),
+            ("integration/x", "refs/tags/v1", self.source),
+            ("integration/x", "feature/x", "z" * 40),
+        ):
+            with self.subTest(integration=integration, source=source), self.assertRaisesRegex(self.m.MergeError, "scope_rejected"):
+                writer.merge(self.m.MergeRequest(integration, source, oid))
+
+    def test_merge_conflict_aborts_and_restores_clean_worktree(self):
+        subprocess.run(["git", "checkout", "-b", "feature/conflict", "integration/x"], cwd=self.work, check=True, capture_output=True)
+        (self.work / "a").write_text("source conflict")
+        subprocess.run(["git", "commit", "-am", "source conflict"], cwd=self.work, check=True, capture_output=True)
+        source = run(self.work, "rev-parse", "HEAD")
+        subprocess.run(["git", "checkout", "integration/x"], cwd=self.work, check=True, capture_output=True)
+        (self.work / "a").write_text("target conflict")
+        subprocess.run(["git", "commit", "-am", "target conflict"], cwd=self.work, check=True, capture_output=True)
+        subprocess.run(["git", "push", "origin", "integration/x"], cwd=self.work, check=True, capture_output=True)
+        before = run(self.work, "rev-parse", "HEAD")
+        remote_head = lambda branch: run(self.work, "ls-remote", "--heads", "origin", "refs/heads/" + branch).split()[0]
+        writer = self.m.IntegrationMergeWriter(self.work, "origin", remote_head)
+        with self.assertRaisesRegex(self.m.MergeError, "integration_merge_conflict"):
+            writer.merge(self.m.MergeRequest("integration/x", "feature/conflict", source))
+        self.assertEqual(before, run(self.work, "rev-parse", "HEAD"))
+        self.assertEqual("", run(self.work, "status", "--porcelain=v1"))
+
+    def test_push_failure_restores_or_enters_reconciliation(self):
+        before = run(self.work, "rev-parse", "HEAD")
+        writer = self.m.IntegrationMergeWriter(self.work, str(self.root / "missing.git"), lambda branch: before)
+        restored = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
+        self.assertEqual("PUSH_FAILED_RESTORED", restored["state"])
+        self.assertEqual(before, run(self.work, "rev-parse", "HEAD"))
+
+        calls = 0
+        def uncertain(branch):
+            nonlocal calls
+            calls += 1
+            if calls == 1: return before
+            raise TimeoutError()
+        writer = self.m.IntegrationMergeWriter(self.work, "origin", uncertain)
+        unknown = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
+        self.assertEqual("RECONCILIATION_REQUIRED", unknown["state"])
+
+
+if __name__ == "__main__": unittest.main()

tests/test_isolated_writer_token_policy_v2.py

@@ -156,6 +156,25 @@ class RepositoryCase(unittest.TestCase):
 
 
 class TokenPolicyV2Tests(RepositoryCase):
+    def test_integration_refresh_grant_is_exactly_bound_and_rotation_revokes_old_token(self):
+        import importlib.util
+        import sys
+        module_path = Path(__file__).parents[1] / "active-runtime/app/aios_isolated_writer_api/integration_refresh.py"
+        spec = importlib.util.spec_from_file_location("integration_refresh_token_test", module_path)
+        refresh = importlib.util.module_from_spec(spec)
+        sys.modules[spec.name] = refresh
+        spec.loader.exec_module(refresh)
+        with tempfile.TemporaryDirectory() as root:
+            store = refresh.GrantStore(Path(root) / "grants.sqlite")
+            binding = refresh.Binding(TASK_ID, "aios/aios", BRANCH, "actor", AUDIENCE, (TARGET_PATH,), 1)
+            old = store.issue(binding)
+            new = store.rotate(old, binding, revocation_probe=lambda token: 401)
+            with self.assertRaisesRegex(refresh.ProvisioningError, "credential_rejected"):
+                store.consume(old, binding, TARGET_PATH)
+            store.consume(new, binding, TARGET_PATH)
+            with self.assertRaisesRegex(refresh.ProvisioningError, "grant_quota_exhausted"):
+                store.consume(new, binding, TARGET_PATH)
+
     def test_v1_coarse_scope_token_is_rejected(self):
         token, record = self.issue()
         record["schema_version"] = 1