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

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