[verified] Add ordered initial rework review input gate

This commit is contained in:
2026-10-12 04:54:43 +00:00
parent aeab04289d
commit b270c2ce30
8 changed files with 722 additions and 7 deletions
@@ -174,10 +174,20 @@ def main():
live.add_argument('--baseline', required=True)
live.add_argument('--baseline-sha256', required=True)
live.add_argument('--action', choices=('create', 'reuse'), required=True)
gate = sp.add_parser('input-gate')
gate.add_argument('--target', required=True)
gate.add_argument('--target-sha256', required=True)
gate.add_argument('--provider-contract', required=True)
gate.add_argument('--provider-contract-sha256', required=True)
gate.add_argument('--actor-id', required=True, type=int)
gate.add_argument('--output', required=True)
args=p.parse_args()
try:
if args.cmd == 'render': do_render(args)
elif args.cmd == 'verify': do_verify(args)
elif args.cmd == 'input-gate':
from rework_readback_cli import run_input_gate
run_input_gate(args)
else:
from rework_readback_cli import run_live
run_live(args, verify_readbacks)
@@ -0,0 +1,236 @@
"""Ordered, conservative review decision gate for rework input."""
from __future__ import annotations
import copy
import hashlib
import json
import re
from datetime import datetime, timezone
SCHEMA = 'gitea-agent.rework-input/v1'
RECEIPT_SCHEMA = 'gitea-agent.rework-input-gate/v1'
ROLES = ('gitea-spec-agent', 'gitea-review-agent')
NAME_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$')
SHA_RE = re.compile(r'^[0-9a-f]{40}$')
DIGEST_RE = re.compile(r'^[0-9a-f]{64}$')
TIME_RE = re.compile(r'^([0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2})(?:\.([0-9]{1,9}))?(Z|[+-][0-9]{2}:[0-9]{2})$')
class InputGateError(ValueError):
pass
def _obj(v, msg='object required'):
if type(v) is not dict: raise InputGateError(msg)
return v
def _id(v, name):
if type(v) is not int or v < 1: raise InputGateError('invalid '+name)
return v
def _sha(v, name, pattern):
if type(v) is not str or not pattern.fullmatch(v): raise InputGateError('invalid '+name)
return v
def digest(v):
try:
raw = json.dumps(v, sort_keys=True, ensure_ascii=False, allow_nan=False, separators=(',', ':')).encode('utf-8')
except (TypeError, ValueError, UnicodeError):
raise InputGateError('invalid JSON artifact') from None
return hashlib.sha256(raw).hexdigest()
def validate_input(value):
value = _obj(value)
if set(value) != {'schema','target','head','contract_sha256'} or value.get('schema') != SCHEMA:
raise InputGateError('invalid rework input schema')
target = _obj(value.get('target'))
if set(target) != {'owner','repository','issue','pr'}: raise InputGateError('invalid target fields')
for name in ('owner','repository'):
if type(target[name]) is not str or not NAME_RE.fullmatch(target[name]): raise InputGateError('invalid target '+name)
_id(target['issue'], 'issue'); _id(target['pr'], 'pr')
if target['issue'] == target['pr']: raise InputGateError('source Issue must differ from PR')
_sha(value.get('head'), 'head', SHA_RE)
_sha(value.get('contract_sha256'), 'contract_sha256', DIGEST_RE)
return copy.deepcopy(value)
def _time_key(value):
m = TIME_RE.fullmatch(value)
if not m: raise InputGateError('invalid submitted_at')
try:
base = datetime.fromisoformat(m.group(1)).replace(tzinfo=timezone.utc)
frac = int((m.group(2) or '').ljust(9, '0'))
zone = m.group(3)
if zone != 'Z':
if int(zone[1:3]) > 23 or int(zone[4:]) > 59:
raise InputGateError('invalid submitted_at offset')
sign = 1 if zone[0] == '+' else -1
off = int(zone[1:3])*3600 + int(zone[4:])*60
base = base.replace(tzinfo=timezone(sign*__import__('datetime').timedelta(seconds=off)))
epoch = base.astimezone(timezone.utc)
return (int(epoch.timestamp()), frac)
except (ValueError, OverflowError):
raise InputGateError('invalid submitted_at') from None
def _marker_fields(body, prefix):
lines = body.splitlines()
matches = [i for i, line in enumerate(lines) if line.startswith(prefix)]
if body.count(prefix) != 1 or len(matches) != 1: raise InputGateError('review marker missing or duplicate')
index = matches[0]
if prefix.endswith('code-review/v1 '):
if index != 0: raise InputGateError('code review marker must be first line')
marker = lines[index]
pattern = (r'<!-- gitea-agent:code-review/v1 owner=([^ ]+) repository=([^ ]+) issue=([1-9][0-9]*) '
r'pr=([1-9][0-9]*) head=([0-9a-f]{40}) contract-sha256=([0-9a-f]{64}) -->')
m = re.fullmatch(pattern, marker)
if not m: raise InputGateError('malformed code review marker')
verdicts = [x for x in lines if x.startswith('- Verdict:')]
if len(verdicts) != 1 or verdicts[0] not in ('- Verdict: `approve`','- Verdict: `request_changes`'):
raise InputGateError('invalid code review verdict')
return {'owner':m.group(1),'repository':m.group(2),'issue':int(m.group(3)),'pr':int(m.group(4)),
'head':m.group(5),'contract_sha256':m.group(6),'verdict':verdicts[0][len('- Verdict: `'):-1],
'marker':marker}
if index != len(lines)-1: raise InputGateError('spec review marker must be last line')
marker = lines[index]
pattern = (r'<!-- gitea-agent:implementation-spec-review/v1 owner=([^ ]+) repository=([^ ]+) issue=([1-9][0-9]*) '
r'pr=([1-9][0-9]*) head=([0-9a-f]{40}) contract-sha256=([0-9a-f]{64}) result-sha256=([0-9a-f]{64}) correlation=([0-9a-f]{64}) -->')
m = re.fullmatch(pattern, marker)
if not m: raise InputGateError('malformed spec review marker')
if not body.endswith('\n'): raise InputGateError('invalid spec review body')
marker_free = body[:body.rfind(marker)]
# The canonical spec renderer adds one separator LF after its hashed body.
if not marker_free.endswith('\n'): raise InputGateError('invalid spec review separator')
result_sha = hashlib.sha256(marker_free[:-1].encode('utf-8')).hexdigest()
if result_sha != m.group(7): raise InputGateError('spec review result digest mismatch')
corr = hashlib.sha256('\0'.join([m.group(1),m.group(2),m.group(3),m.group(4),m.group(5),m.group(6),m.group(7)]).encode()).hexdigest()
if corr != m.group(8): raise InputGateError('spec review correlation mismatch')
verdicts = [x for x in lines if x.startswith('- Verdict:')]
if len(verdicts) != 1 or verdicts[0] not in ('- Verdict: `approve`','- Verdict: `request_changes`','- Verdict: `blocked_spec`'):
raise InputGateError('invalid spec review verdict')
return {'owner':m.group(1),'repository':m.group(2),'issue':int(m.group(3)),'pr':int(m.group(4)),
'head':m.group(5),'contract_sha256':m.group(6),'verdict':verdicts[0][len('- Verdict: `'):-1],
'marker':marker}
def _validate_raw(adapter, raw, inp):
target = inp['target']; prefix = '/repos/'+target['owner']+'/'+target['repository']
# Existing canonical guards validate native metadata, endpoint identity, counts and bindings.
target = dict(target, head=inp['head'])
adapter.bind_input(target, raw, require_default_branch=True)
adapter.validate_collection(raw['source_comments'], prefix+'/issues/'+str(target['issue'])+'/comments')
adapter.validate_collection(raw['comments'], prefix+'/issues/'+str(target['pr'])+'/comments')
adapter.validate_collection(raw['reviews'], prefix+'/pulls/'+str(target['pr'])+'/reviews', reviews=True)
expected_inline = {str(row['id']) for row in raw['reviews']['rows']}
if set(raw.get('inline', {})) != expected_inline:
raise InputGateError('incomplete review comment endpoints')
for rid, collected in raw['inline'].items():
adapter.validate_collection(collected, prefix+'/pulls/'+str(target['pr'])+'/reviews/'+rid+'/comments',
allow_missing=adapter.provider['inline']['mode'] == 'per_review_or_pr_count')
adapter.counts(raw)
source = raw['source_issue']
if source['body'].splitlines()[:1] != ['<!-- gitea-agent:discovery-draft/v2 -->']:
raise InputGateError('invalid discovery draft marker')
if hashlib.sha256(source['body'].encode('utf-8')).hexdigest() != inp['contract_sha256']:
raise InputGateError('Source Issue contract digest mismatch')
if raw['pr']['head']['sha'] != inp['head']: raise InputGateError('PR head mismatch')
return prefix
def derive_decision(raw, inp, impl_actor_id):
inp = validate_input(inp); _id(impl_actor_id, 'actor_id')
timeline=[]; role_ids={}
for review in raw['reviews']['rows']:
user = _obj(review.get('user')); login = user.get('login'); rid = _id(review.get('id'),'review id')
role = login if login in ROLES else 'unknown'
if login in ROLES:
uid = _id(user.get('id'),'reviewer id')
if uid == impl_actor_id: raise InputGateError('reviewer identity equals implementation actor')
prior = role_ids.get(login)
if prior is not None and prior != uid: raise InputGateError('inconsistent reviewer identity')
if prior is None and uid in role_ids.values(): raise InputGateError('reviewer roles share identity')
role_ids[login] = uid
marker = None; marker_error = None
try:
if role in ROLES: marker = _marker_fields(review.get('body',''), '<!-- gitea-agent:'+('implementation-spec-review/v1 ' if role == 'gitea-spec-agent' else 'code-review/v1 '))
except InputGateError as exc: marker_error = str(exc)
if marker is not None:
t=inp['target']
if (marker['owner'],marker['repository'],marker['issue'],marker['pr']) != (t['owner'],t['repository'],t['issue'],t['pr']) or marker['head'] != review['commit_id'] or marker['contract_sha256'] != inp['contract_sha256']:
marker_error='review marker binding mismatch'
current = review['commit_id'] == inp['head']
reason=[]
if not current: reason.append('old_head')
if review.get('stale'): reason.append('stale')
if review.get('dismissed'): reason.append('dismissed')
if marker_error: reason.append('malformed_marker' if 'marker' in marker_error or 'verdict' in marker_error or 'digest' in marker_error or 'correlation' in marker_error else 'marker_binding')
if role == 'unknown': reason.append('unknown_role')
if review['state'] == 'PENDING': reason.append('pending')
if review['state'] == 'COMMENT': reason.append('comment_manual_classification')
if marker is not None:
compatible = {'approve': 'APPROVED', 'request_changes': 'REQUEST_CHANGES',
'blocked_spec': 'COMMENT'}[marker['verdict']]
if review['state'] != compatible:
reason.append('state_marker_contradiction')
if role in ROLES and current and marker_error: reason.append('authoritative_marker_error')
timeline.append({'id':rid,'author':{'id':user['id'],'login':login,'role':role},'state':review['state'],
'submitted_at':review['submitted_at'],'reviewed_sha':review['commit_id'],'stale':review['stale'],'dismissed':review['dismissed'],
'canonical_body_marker': marker['marker'] if marker else None,'body_sha256':hashlib.sha256(review['body'].encode()).hexdigest(),
'reason_codes':reason,'eligible':role in ROLES and current and review['state'] != 'COMMENT' and not review['stale'] and not review['dismissed'] and marker is not None and marker_error is None and 'state_marker_contradiction' not in reason,
'_time':_time_key(review['submitted_at']),'_verdict':marker['verdict'] if marker else None,'_role':role,'_marker_error':marker_error})
if len(timeline) > 1000: raise InputGateError('decision artifact too large')
timeline.sort(key=lambda x:(x['_time'],x['id']))
reasons=[]; status='blocked'; selected={}
if any('authoritative_marker_error' in x['reason_codes'] for x in timeline): reasons.append('authoritative_marker_error')
if any('state_marker_contradiction' in x['reason_codes'] and x['reviewed_sha'] == inp['head'] for x in timeline): reasons.append('state_marker_contradiction')
if any(x['_role'] in ROLES and x['reviewed_sha'] == inp['head'] and x['state'] == 'COMMENT' and not x['stale'] and not x['dismissed'] for x in timeline):
reasons.append('comment_manual_classification')
if any(x['_role'] == 'gitea-spec-agent' and x['reviewed_sha'] == inp['head']
and not x['stale'] and not x['dismissed'] and x['_verdict'] == 'blocked_spec' for x in timeline):
reasons.append('blocked_spec')
for role in ROLES:
all_current = [x for x in timeline if x['_role'] == role and x['reviewed_sha'] == inp['head']
and not x['stale'] and not x['dismissed']]
by_time = {}
for row in all_current:
by_time.setdefault(row['_time'], []).append(row)
conflicting_role = False
for same_time in by_time.values():
signatures = {(x['state'], x['body_sha256'], x.get('canonical_body_marker')) for x in same_time}
if len(signatures) > 1:
reasons.append('conflicting_same_time_reviews')
selected[role] = None
conflicting_role = True
break
if conflicting_role:
continue
rows=[x for x in timeline if x['_role']==role and x['eligible']]
if any(x['state']=='PENDING' and x['_role']==role and x['reviewed_sha']==inp['head'] and not x['stale'] and not x['dismissed'] for x in timeline):
reasons.append('pending_authoritative_review')
if not rows:
selected[role]=None; reasons.append('missing_current_review')
continue
latest_time=rows[-1]['_time']; same=[x for x in rows if x['_time']==latest_time]
verdicts={x['_verdict'] for x in same}
if len(verdicts)>1:
reasons.append('conflicting_same_time_reviews'); selected[role]=None; continue
latest=same[-1]; selected[role]=latest['id']
if latest['_verdict']=='request_changes': reasons.append('request_changes')
elif latest['_verdict']=='blocked_spec': reasons.append('blocked_spec')
if not reasons:
status='no_op'
elif reasons and set(reasons) == {'request_changes'}:
status='rework'
else: status='blocked'
for row in timeline:
for key in ('_time','_verdict','_role','_marker_error'): row.pop(key,None)
decision={'status':status,'reason_codes':sorted(set(reasons)),'target':copy.deepcopy(inp['target']),'head':inp['head'],
'contract_sha256':inp['contract_sha256'],'actor':{'id':impl_actor_id,'login':'gitea-impl-agent','role':'gitea-impl-agent'},
'selected_review_ids':selected,'timeline':timeline,
'rework_review_ids':[row['id'] for row in timeline if status == 'rework'
and row['id'] in selected.values() and row['state'] == 'REQUEST_CHANGES']}
if len(json.dumps(decision,ensure_ascii=False,sort_keys=True).encode()) > 1_000_000: raise InputGateError('decision artifact too large')
return decision
def build_receipt(raw, inp, decision):
receipt = {'schema':RECEIPT_SCHEMA,'status':decision['status'],'reason_codes':decision['reason_codes'],
'target':decision['target'],'head':decision['head'],'contract_sha256':decision['contract_sha256'],
'actor':decision['actor'],'input_sha256':digest(inp),'raw_input_snapshot':copy.deepcopy(raw),
'raw_input_sha256':digest(raw),'fresh_GET_provenance':{'kind':'fresh_authenticated_GET','read_only':True},
'decision':decision}
if len(json.dumps(receipt, ensure_ascii=False, sort_keys=True, allow_nan=False).encode('utf-8')) > 1_000_000:
raise InputGateError('receipt artifact too large')
return receipt
@@ -276,7 +276,7 @@ class ReworkReadbackAdapter:
raise ReadbackError('expected renderer binding mismatch')
return dict(target, review=expected['review'], reviewed_head=expected['reviewed_head'], head=expected['head'])
def bind(self, target, raw):
def bind_input(self, target, raw, *, require_default_branch=False):
prefix = ORIGIN + '/repos/' + target['owner'] + '/' + target['repository']
html = 'https://gitea.jsfamily.quest/' + target['owner'] + '/' + target['repository']
repo = object_value(raw['repository'])
@@ -310,12 +310,20 @@ class ReworkReadbackAdapter:
raise ReadbackError('PR base repository mismatch')
if head_branch['sha'] != target['head']:
raise ReadbackError('PR head mismatch')
if require_default_branch:
default_branch = repo.get('default_branch')
if type(default_branch) is not str or not default_branch or any(ord(c) < 33 for c in default_branch):
raise ReadbackError('repository default branch missing')
if base_branch['ref'] != default_branch:
raise ReadbackError('PR base default branch mismatch')
# Fork heads are valid; the target repository is bound through base and endpoint.
relation = object_value(underlying.get('pull_request'))
if (underlying.get('html_url') not in (html+'/pulls/'+str(target['pr']), html+'/issues/'+str(target['pr']))
or relation.get('html_url') != html+'/pulls/'+str(target['pr'])
or ('url' in relation and relation.get('url') != html+'/pulls/'+str(target['pr']))):
raise ReadbackError('underlying Issue PR binding mismatch')
def bind(self, target, raw):
self.bind_input(target, raw)
review = next((row for row in raw['reviews']['rows'] if row.get('id') == target['review']), None)
if review is None or review.get('commit_id') != target['reviewed_head']:
raise ReadbackError('source review binding mismatch')
@@ -359,7 +367,7 @@ class ReworkReadbackAdapter:
raise ReadbackError('source collection count mismatch')
return counts
def capture(self, target):
def capture_input(self, target, *, require_default_branch=True):
identity, _ = self.get('/user')
if integer(identity.get('id'), 1) != self.actor_id or identity.get('login') != 'gitea-impl-agent':
raise ReadbackError('authenticated identity mismatch')
@@ -378,10 +386,36 @@ class ReworkReadbackAdapter:
raw[name] = self.collection(endpoint)
raw['inline'] = {str(review['id']):self.collection(prefix+'/pulls/'+str(target['pr'])+
'/reviews/'+str(review['id'])+'/comments', allow_missing=self.provider['inline']['mode'] == 'per_review_or_pr_count') for review in raw['reviews']['rows']}
self.bind(target, raw)
self.bind_input(target, raw, require_default_branch=require_default_branch)
self.counts(raw)
return raw
def capture(self, target):
raw = self.capture_input(target, require_default_branch=False)
self.bind(target, raw)
return raw
def input_snapshot(self, value):
import importlib.util, sys
path = __import__('pathlib').Path(__file__).with_name('rework_input_gate.py')
if 'rework_input_gate' not in sys.modules:
spec = importlib.util.spec_from_file_location('rework_input_gate', path)
if spec is None or spec.loader is None: raise ReadbackError('input gate unavailable')
gate = importlib.util.module_from_spec(spec); sys.modules[spec.name] = gate; spec.loader.exec_module(gate)
else:
gate = sys.modules['rework_input_gate']
build_receipt, derive_decision, digest, validate_input = gate.build_receipt, gate.derive_decision, gate.digest, gate.validate_input
value = validate_input(value)
target = dict(value['target'], head=value['head'], contract_sha256=value['contract_sha256'])
first = self.capture_input(target)
gate._validate_raw(self, first, value)
second = self.capture_input(target)
gate._validate_raw(self, second, value)
if digest(first) != digest(second):
raise ReadbackError('concurrent provider drift')
decision = derive_decision(first, value, self.actor_id)
return build_receipt(first, value, decision)
def snapshot(self, expected):
target = self.target(expected)
raw = self.capture(target)
@@ -191,8 +191,9 @@ def read_bound_json(path, expected_sha256=None):
class OutputFile:
"""Pin a private output directory, and replace only after successful verification."""
def __init__(self, path):
def __init__(self, path, *, initial_only=False):
self.path = Path(path)
self.initial_only = initial_only
self.fd, self.name = _open_parent(self.path)
try:
parent = os.fstat(self.fd)
@@ -200,6 +201,8 @@ class OutputFile:
raise ValueError('private output directory required')
self.parent_identity = (parent.st_dev, parent.st_ino)
self.original = self.identity()
if self.initial_only and self.original is not None:
raise ValueError('output already exists')
self.done = False
except BaseException:
os.close(self.fd)
@@ -225,7 +228,7 @@ class OutputFile:
def write(self, value):
raw = (json.dumps(value, ensure_ascii=False, sort_keys=True, allow_nan=False) + '\n').encode()
if len(raw) > 16_000_000 or self.done:
if len(raw) > (1_000_000 if self.initial_only else 16_000_000) or self.done:
raise ValueError('invalid output size or repeated write')
self.check_parent()
if self.identity() != self.original:
@@ -242,7 +245,15 @@ class OutputFile:
self.check_parent()
if self.identity() != self.original:
raise ValueError('output target drift')
os.replace(temporary, self.name, src_dir_fd=self.fd, dst_dir_fd=self.fd)
if self.initial_only:
try:
os.link(temporary, self.name, src_dir_fd=self.fd, dst_dir_fd=self.fd,
follow_symlinks=False)
except FileExistsError:
raise ValueError('output created during capture') from None
os.unlink(temporary, dir_fd=self.fd)
else:
os.replace(temporary, self.name, src_dir_fd=self.fd, dst_dir_fd=self.fd)
self.done = True
os.fsync(self.fd)
self.check_parent()
@@ -256,6 +267,21 @@ class OutputFile:
def __exit__(self, *args): os.close(self.fd)
def reject_existing_output(path):
"""Initial gate artifacts are immutable run-specific outputs."""
parent, name = _open_parent(path)
try:
try:
metadata = os.stat(name, dir_fd=parent, follow_symlinks=False)
except FileNotFoundError:
return
if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or metadata.st_uid != os.getuid():
raise ValueError('unsafe existing output target')
raise ValueError('output already exists')
finally:
os.close(parent)
def run_live(args, verifier):
"""Capture a baseline or feed computed fresh readbacks into the existing verifier."""
try:
@@ -316,3 +342,41 @@ def run_live(args, verifier):
print(json.dumps({'output': str(args.output), 'sha256': digest, 'read_only': True}))
except Exception:
raise ValueError('live readback failed; output not verified') from None
def run_input_gate(args):
"""Capture and decide initial input without overwriting an existing receipt."""
input_paths = [Path(args.target), Path(args.provider_contract)]
if Path(args.output) in input_paths:
raise ValueError('output aliases protected input')
with OutputFile(args.output, initial_only=True) as output:
target = read_bound_json(args.target, args.target_sha256)
contract = read_bound_json(args.provider_contract, args.provider_contract_sha256)
import importlib.util, sys
gate_path = Path(__file__).with_name('rework_input_gate.py')
gate_spec = importlib.util.spec_from_file_location('rework_input_gate', gate_path)
if gate_spec is None or gate_spec.loader is None: raise ValueError('input gate unavailable')
gate = importlib.util.module_from_spec(gate_spec); sys.modules[gate_spec.name] = gate; gate_spec.loader.exec_module(gate)
target = gate.validate_input(target)
core_path = Path(__file__).with_name('rework_readback.py')
core_spec = importlib.util.spec_from_file_location('gitea_input_preflight', core_path)
if core_spec is None or core_spec.loader is None: raise ValueError('canonical readback core unavailable')
core = importlib.util.module_from_spec(core_spec); sys.modules[core_spec.name] = core; core_spec.loader.exec_module(core)
core.ReworkReadbackAdapter(lambda path: (_ for _ in ()).throw(ValueError('preflight transport must not run')),
origin=ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN',
actor_id=args.actor_id, provider_contract=contract)
home = os.environ.get('HERMES_HOME')
if not home:
raise ValueError('role profile not bound')
credential = load_role_token(home)
transport = RestTransport(credential)
path = Path(__file__).with_name('rework_readback.py')
spec = importlib.util.spec_from_file_location('gitea_impl_readback_core', path)
if spec is None or spec.loader is None: raise ValueError('canonical readback core unavailable')
module = importlib.util.module_from_spec(spec); sys.modules[spec.name] = module; spec.loader.exec_module(module)
adapter = module.ReworkReadbackAdapter(transport, origin=ORIGIN + '/api/v1',
credential_source='$HERMES_HOME/.env:GITEA_TOKEN', actor_id=args.actor_id, provider_contract=contract)
result = adapter.input_snapshot(target)
if credential in json.dumps(result, ensure_ascii=False): raise ValueError('credential material in artifact')
output.write(result)
print(json.dumps({'output':str(args.output), 'read_only':True, 'immutable':True}))