270 lines
14 KiB
Python
270 lines
14 KiB
Python
"""Root-executed, hash-bound draft/upload/publication. Import has no side effects."""
|
|
if not __debug__:
|
|
raise RuntimeError('Publication requires Python assertions; optimized execution is refused.')
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import tomllib
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
ORIGIN = 'http://217.177.34.108:3000'
|
|
API = ORIGIN + '/api/v1/repos/sorokin.oleg.edu/ARCADIA-Studio'
|
|
TAG = 'v0.2.0-preview.rename.20260912'
|
|
TARGET = 'cbc49a7802fb1bba0dd284fd904c700a0703ae70'
|
|
TITLE = 'Source alias rename — Windows development preview'
|
|
ZIP = 'ArcadiaStudio-0.2.0-preview.rename.20260912-x86_64-pc-windows-gnullvm.zip'
|
|
ZIP_SHA = 'd2433b5b384256b551672bf115bab40b3496aeee115e067922aedb5becf34d7e'
|
|
CHECKSUM_SHA = '3b8d811c73ae0044ed0b451e8ff746f440af79b543dd52fe067a5a2e175b685f'
|
|
|
|
def now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
def digest(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
def read(path):
|
|
return json.loads(path.read_text(encoding='utf-8'))
|
|
|
|
def exclusive(path, value):
|
|
with path.open('x', encoding='utf-8', newline='\n') as stream:
|
|
json.dump(value, stream, indent=2, ensure_ascii=False)
|
|
stream.write('\n')
|
|
|
|
class Attempt:
|
|
def __init__(self, action, name):
|
|
assert re.fullmatch(r'[a-z0-9][a-z0-9-]{0,47}', name), 'Use a fresh simple attempt name'
|
|
self.path = HERE / 'runs' / (action + '-' + name)
|
|
self.path.mkdir(parents=True, exist_ok=False)
|
|
self.value = {'started': now(), 'action': action, 'tag': TAG, 'target': TARGET, 'success': False}
|
|
|
|
def event(self, step, **fields):
|
|
event = {'atUtc': now(), 'step': step, **fields}
|
|
with (self.path / 'events.jsonl').open('a', encoding='utf-8', newline='\n') as stream:
|
|
stream.write(json.dumps(event, ensure_ascii=False) + '\n')
|
|
|
|
def finish(self):
|
|
self.value['finished'] = now()
|
|
exclusive(self.path / 'result.json', self.value)
|
|
print(json.dumps(self.value, ensure_ascii=False), flush=True)
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
raise RuntimeError('Unexpected redirect; authorization not forwarded')
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
config = tomllib.loads((Path.home() / '.codex/config.toml').read_text(encoding='utf-8'))
|
|
self.token = config['mcp_servers']['gitea']['env']['GITEA_ACCESS_TOKEN']
|
|
self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect)
|
|
|
|
def redact(self, text):
|
|
return text.replace(self.token, '[redacted]')
|
|
|
|
def request(self, url, data=None, method=None, headers=None, binary=False, missing=False):
|
|
parsed = urllib.parse.urlsplit(url)
|
|
expected = urllib.parse.urlsplit(ORIGIN)
|
|
assert (parsed.scheme, parsed.netloc) == (expected.scheme, expected.netloc)
|
|
assert parsed.username is None and parsed.password is None
|
|
request = urllib.request.Request(url, data=data, method=method,
|
|
headers={'Authorization': 'token ' + self.token, **(headers or {})})
|
|
try:
|
|
with self.opener.open(request, timeout=120) as response:
|
|
body = response.read()
|
|
return body if binary else json.loads(body)
|
|
except urllib.error.HTTPError as error:
|
|
detail = self.redact(error.read(2000).decode('utf-8', 'replace'))
|
|
if missing and error.code == 404:
|
|
return None
|
|
raise RuntimeError(f'HTTP {error.code}: {detail}') from None
|
|
|
|
def json(self, url, value, method):
|
|
return self.request(url, json.dumps(value).encode('utf-8'), method,
|
|
{'Content-Type': 'application/json'})
|
|
|
|
def release(self):
|
|
# Listing authenticated drafts catches a reserved-tag collision even
|
|
# before a server has made its Git tag visible to the by-tag endpoint.
|
|
matches = []
|
|
for page in range(1, 21):
|
|
items = self.request(API + f'/releases?limit=50&page={page}')
|
|
assert isinstance(items, list)
|
|
matches.extend(item for item in items if item['tag_name'] == TAG)
|
|
if len(items) < 50:
|
|
break
|
|
else:
|
|
raise RuntimeError('Release collision scan exceeded its bound')
|
|
assert len(matches) <= 1, 'Multiple releases claim the reserved tag'
|
|
return matches[0] if matches else None
|
|
|
|
def tag(self):
|
|
return self.request(API + '/tags/' + urllib.parse.quote(TAG, safe=''), missing=True)
|
|
|
|
def inputs():
|
|
bindings = read(HERE / 'bindings.json')
|
|
assert bindings['acceptanceReviewed'] is True, 'Parent must finish and explicitly review actual acceptance first'
|
|
assert isinstance(bindings['sourceBranch'], str) and bindings['sourceBranch']
|
|
assert re.fullmatch(r'[0-9a-f]{40}', bindings['sourceBranchHead'] or ''), 'Pin current published branch head separately from release target'
|
|
assert set(bindings['acceptanceEvidence']) == {'sourceBinding', 'packageIntegrity', 'packagedServer', 'nativeClosure'}
|
|
for key, item in bindings['acceptanceEvidence'].items():
|
|
assert item['reviewedPass'] is True, 'Missing parent acceptance review: ' + key
|
|
assert isinstance(item['sha256'], str) and re.fullmatch(r'[0-9a-f]{64}', item['sha256']), 'Missing acceptance receipt pin: ' + key
|
|
path = Path(item['path'] or '')
|
|
assert path.is_absolute() and path.is_file(), 'Missing acceptance receipt: ' + key
|
|
raw = path.read_bytes()
|
|
assert len(raw) == item['bytes'] and digest(raw) == item['sha256'], 'Acceptance receipt changed: ' + key
|
|
for key in ('notesSha256', 'validationSha256'):
|
|
assert isinstance(bindings[key], str) and re.fullmatch(r'[0-9a-f]{64}', bindings[key]), 'Final notes and VALIDATION hashes must be reviewed and bound first'
|
|
notes = (HERE / 'RELEASE-NOTES.md').read_bytes()
|
|
validation = (HERE / 'VALIDATION.md').read_bytes()
|
|
assert digest(notes) == bindings['notesSha256']
|
|
assert digest(validation) == bindings['validationSha256']
|
|
for raw in (notes, validation):
|
|
text = raw.decode('utf-8')
|
|
assert all(marker not in text for marker in ('PUBLICATION-PENDING', 'PACKAGE-RUNTIME-ACCEPTANCE-PENDING'))
|
|
package = Path(bindings['packageDirectory'])
|
|
assert package.is_absolute() and package.is_dir()
|
|
assets = {ZIP: (package / ZIP).read_bytes(), ZIP + '.sha256': (package / (ZIP + '.sha256')).read_bytes(), 'VALIDATION.md': validation}
|
|
assert len(assets[ZIP]) == 33074937 and digest(assets[ZIP]) == ZIP_SHA
|
|
assert digest(assets[ZIP + '.sha256']) == CHECKSUM_SHA
|
|
assert assets[ZIP + '.sha256'].decode('utf-8').strip() == ZIP_SHA + ' ' + ZIP
|
|
return bindings, notes.decode('utf-8'), assets
|
|
|
|
def source_and_tag(client, require_tag=True):
|
|
bindings = read(HERE / 'bindings.json')
|
|
branch = client.request(API + '/branches/' + urllib.parse.quote(bindings['sourceBranch'], safe=''))
|
|
assert branch['commit']['id'] == bindings['sourceBranchHead'], 'Published source branch changed'
|
|
target_commit = client.request(API + '/git/commits/' + TARGET)
|
|
assert target_commit['sha'] == TARGET, 'Exact source snapshot is unavailable'
|
|
tag = client.tag()
|
|
if tag is None:
|
|
assert not require_tag, 'Parent must create the reviewed exact-target tag through Gitea MCP first'
|
|
else:
|
|
assert tag['name'] == TAG and tag['commit']['sha'] == TARGET, 'Tag collision'
|
|
return tag
|
|
|
|
def owned_release(value, bindings, notes):
|
|
expected_id = bindings.get('releaseId')
|
|
state = HERE / 'release-created.json'
|
|
if state.exists():
|
|
saved = read(state)
|
|
assert saved['tag'] == TAG and saved['target'] == TARGET
|
|
assert expected_id in (None, saved['releaseId'])
|
|
expected_id = saved['releaseId']
|
|
assert expected_id is not None and value['id'] == expected_id, 'Existing release is not bound to this run; inspect and review before setting releaseId'
|
|
assert value['tag_name'] == TAG and value['target_commitish'] == TARGET
|
|
assert value['prerelease'] is True and value['name'] == TITLE and value['body'] == notes
|
|
|
|
def verify_assets(client, release_id, assets, allow_missing=False):
|
|
remote = client.request(API + f'/releases/{release_id}/assets')
|
|
names = [item['name'] for item in remote]
|
|
assert len(names) == len(set(names)) and len({a['id'] for a in remote}) == len(remote)
|
|
assert set(names) <= set(assets)
|
|
if not allow_missing:
|
|
assert set(names) == set(assets), 'Exactly three assets required'
|
|
verified = []
|
|
for item in remote:
|
|
expected = assets[item['name']]
|
|
assert item['size'] == len(expected)
|
|
received = client.request(item['browser_download_url'], binary=True)
|
|
assert received == expected, 'Existing asset bytes differ; no overwrite or deletion permitted'
|
|
verified.append({'id': item['id'], 'name': item['name'], 'bytes': len(received), 'sha256': digest(received), 'url': item['browser_download_url']})
|
|
return verified
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('action', choices=('inspect', 'create', 'upload', 'publish'))
|
|
parser.add_argument('--attempt', required=True)
|
|
args = parser.parse_args()
|
|
attempt = Attempt(args.action, args.attempt)
|
|
client = None
|
|
try:
|
|
if args.action == 'inspect':
|
|
client = Client()
|
|
release, tag = client.release(), client.tag()
|
|
attempt.value.update(success=True, release=release, remoteTag=tag,
|
|
scope='Read-only inspection; no acceptance or mutation')
|
|
return
|
|
bindings, notes, assets = inputs()
|
|
attempt.value['bindings'] = bindings
|
|
attempt.value['assetHashes'] = {name: digest(raw) for name, raw in assets.items()}
|
|
client = Client()
|
|
source_and_tag(client)
|
|
release = client.release()
|
|
if args.action == 'create':
|
|
if release is not None:
|
|
owned_release(release, bindings, notes)
|
|
assert release['draft'] is True
|
|
attempt.event('existing-owned-draft', releaseId=release['id'])
|
|
else:
|
|
assert not (HERE / 'release-created.json').exists() and bindings.get('releaseId') is None
|
|
attempt.event('create-request-started')
|
|
release = client.json(API + '/releases', {'tag_name': TAG, 'target_commitish': TARGET,
|
|
'name': TITLE, 'body': notes, 'draft': True, 'prerelease': True}, 'POST')
|
|
assert release['tag_name'] == TAG and release['target_commitish'] == TARGET and release['draft'] is True
|
|
# Persist ownership immediately; a failed readback can be retried safely.
|
|
exclusive(HERE / 'release-created.json', {'releaseId': release['id'], 'tag': TAG, 'target': TARGET, 'atUtc': now()})
|
|
attempt.event('created-draft', releaseId=release['id'])
|
|
fresh = client.release()
|
|
owned_release(fresh, bindings, notes)
|
|
assert fresh['draft'] is True
|
|
verify_assets(client, fresh['id'], assets, allow_missing=True)
|
|
else:
|
|
assert release is not None, 'Create the hash-bound draft first'
|
|
owned_release(release, bindings, notes)
|
|
endpoint = API + f'/releases/{release["id"]}'
|
|
if args.action == 'upload':
|
|
assert release['draft'] is True, 'Never upload into a public release'
|
|
existing = verify_assets(client, release['id'], assets, allow_missing=True)
|
|
for name, raw in assets.items():
|
|
if any(item['name'] == name for item in existing):
|
|
attempt.event('existing-asset-verified', name=name, sha256=digest(raw))
|
|
continue
|
|
boundary = 'ArcadiaRename' + uuid.uuid4().hex
|
|
prefix = (f'--{boundary}\r\nContent-Disposition: form-data; name="attachment"; filename="{name}"\r\nContent-Type: application/octet-stream\r\n\r\n').encode()
|
|
attempt.event('upload-request-started', name=name, sha256=digest(raw))
|
|
result = client.request(endpoint + '/assets?' + urllib.parse.urlencode({'name': name}),
|
|
prefix + raw + f'\r\n--{boundary}--\r\n'.encode(), 'POST',
|
|
{'Content-Type': 'multipart/form-data; boundary=' + boundary})
|
|
assert result['name'] == name and result['size'] == len(raw)
|
|
assert client.request(result['browser_download_url'], binary=True) == raw
|
|
attempt.event('uploaded-and-downloaded', name=name, assetId=result['id'], sha256=digest(raw))
|
|
attempt.value['assets'] = verify_assets(client, release['id'], assets)
|
|
else:
|
|
# Download all three afresh immediately before changing visibility.
|
|
before = verify_assets(client, release['id'], assets)
|
|
source_and_tag(client)
|
|
fresh = client.release()
|
|
owned_release(fresh, bindings, notes)
|
|
if fresh['draft']:
|
|
attempt.event('publish-request-started', releaseId=fresh['id'])
|
|
client.json(endpoint, {'draft': False, 'prerelease': True, 'body': notes}, 'PATCH')
|
|
final = client.release()
|
|
owned_release(final, bindings, notes)
|
|
assert final['draft'] is False
|
|
source_and_tag(client)
|
|
after = verify_assets(client, final['id'], assets)
|
|
assert sorted(before, key=lambda x: x['id']) == sorted(after, key=lambda x: x['id'])
|
|
attempt.value.update(assets=after, url=final['html_url'], apiPublishedAt=final['published_at'], visibilityVerifiedAt=now())
|
|
fresh = client.release()
|
|
owned_release(fresh, bindings, notes)
|
|
attempt.value.update(success=True, releaseId=fresh['id'], draft=fresh['draft'], prerelease=True)
|
|
except BaseException as error:
|
|
message = str(error)
|
|
attempt.value['error'] = client.redact(message) if client else message
|
|
attempt.value['remoteOutcome'] = 'May be partial if a request started; inspect before retry. No automatic overwrite/delete.'
|
|
raise SystemExit(1) from None
|
|
finally:
|
|
attempt.finish()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|