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-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

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"
+}

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

@@ -1,11 +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.
-
-It intentionally has no live activation entry point. Both systemd services are
-static, refuse manual start, carry no credential, and keep all write and
-activation flags false. A separate `SINGLE_OWNER_GOVERNANCE_ATOMIC_ACTIVATION`
-approval is required before any live status publication or protection change.

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/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/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,13 +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 = []
-
-[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_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,30 +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((ROOT / "deploy" / "systemd").glob("aios-governance-*.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)
-
-
-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