Talondri: Protected "User:EliteMaintenanceBot/Governance source": Preserve reviewed operational bot source ([Edit=Allow only administrators] (indefinite) [Move=Allow only administrators] (indefinite))

2026-11-01T17:08:39Z

Protected "User:EliteMaintenanceBot/Governance source": Preserve reviewed operational bot source ([Edit=Allow only administrators] (indefinite) [Move=Allow only administrators] (indefinite))

← Older revision		Revision as of 17:08, 1 November 2026
(No difference)

Talondri: Publish reviewed secret-free governance worker and authorised task scope

2026-11-01T17:08:35Z

Publish reviewed secret-free governance worker and authorised task scope

New page

Operator: [[User:Talondri|Talondri]] (site operator). This bounded task was explicitly authorised by the site operator. It uses the existing EliteMaintenanceBot account and existing restricted credentials. No new human account access was granted.

The worker transfers complete, explicitly closed level-two discussions after at least 30 days, only with a dated recorded outcome and activity checks. It reports election deadlines from the two open-election categories. It never closes discussions, counts votes, declares results or grants rights. It honours bot exclusions and administrator-only JSON controls.

Code review identified an archive-change race during edit pacing. The corrected worker rechecks the archive after pacing, immediately before source removal. Offline regression tests and isolated live user-subpage lifecycle tests were run. This records the authorised trial and review evidence, not a claim of wider community consensus.

* [[User:EliteMaintenanceBot/Governance control|Administrator control]]
* [[User:EliteMaintenanceBot/Governance report|Deadline report]]
* [[Project:Bots|Local bot policy]]

== Published worker ==
<syntaxhighlight lang="python">
"""Bounded governance housekeeping. Authenticated dry-run unless --apply."""
from __future__ import annotations

import argparse
from contextlib import contextmanager
from datetime import date, datetime, timezone, timedelta
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import tempfile
import time

import mwparserfromhell as mw

BOT = 'EliteMaintenanceBot'
CONTROL = 'User:EliteMaintenanceBot/Governance control'
REPORT = 'User:EliteMaintenanceBot/Governance report'
TEST = 'User:EliteMaintenanceBot/Governance tests/'
BOARDS = {"Project:" + s for s in (
"Administrators' noticeboard", "Bureaucrats' noticeboard", 'Requests for permissions',
'Requests for page protection', 'Bot requests', 'Village pump', 'Help desk')}
CATEGORIES = ('Category:Open administrator elections', 'Category:Open bureaucrat elections')
START, END = '', ''
MAX_BYTES = 500_000

class Halt(RuntimeError):
pass

def name(value):
return str(value).strip().replace('_', ' ').casefold().removeprefix('template:').strip()

def params(template):
result = {}
for p in template.params:
key = name(p.name)
if key in result:
raise Halt('Duplicate template parameters')
result[key] = str(p.value).strip()
return result

def iso(value):
if not isinstance(value, str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}', value):
raise ValueError('Date requires YYYY-MM-DD')
return date.fromisoformat(value)

def optout(text):
for t in mw.parse(text).filter_templates(recursive=True):
n = name(t.name)
if n == 'nobots':
raise Halt('Bot exclusion')
if n == 'bots':
p = params(t)
for key in ('deny', 'allow', 'optout'):
if key not in p:
continue
values = {name(v) for v in p[key].split(',')}
if ((key == 'deny' and values & {'all', BOT.casefold()})
or (key == 'allow' and not values & {'all', BOT.casefold()})
or (key == 'optout' and values & {'all', 'governance', 'archive', 'archiving'})):
raise Halt('Bot exclusion')

def sections(text, today):
"""Only genuine top-level level-2 sections, terminated by level <=2 headings.

Parser nodes avoid headings/markers hidden in comments, nowiki, tags or templates.
The marker must belong directly to the level-2 section, before any subheading.
"""
nodes = mw.parse(text).nodes
positions, offset = [], 0
for i, node in enumerate(nodes):
if isinstance(node, mw.nodes.Heading) and node.level <= 2:
positions.append((i, offset, node.level))
offset += len(str(node))
positions.append((len(nodes), len(text), 0))
eligible = []
for current, following in zip(positions, positions[1:]):
i, begin, level = current
j, end, _ = following
if level != 2:
continue
direct = []
for node in nodes[i+1:j]:
if isinstance(node, mw.nodes.Heading):
break
if isinstance(node, mw.nodes.Template) and name(node.name) == 'governance closed':
direct.append(node)
if len(direct) != 1:
continue
try:
p = params(direct[0])
if (not {'date', 'outcome'} <= set(p) or set(p) - {'date', 'outcome', 'by'}
or not p['outcome'] or len(p['outcome']) > 1000):
continue
if (p['outcome'].casefold() in {'unclear', 'reopened', 'open', 'pending', 'undecided', 'tbd', 'unknown'}
or re.search(r'[{}<>\n\r]', p['outcome'])):
continue
closed = iso(p['date'])
except (ValueError, Halt):
continue
content = text[begin:end]
if any(name(t.name) in ('governance reopened', 'reopened')
for t in mw.parse(content).filter_templates(recursive=True)):
continue
if (today - closed).days >= 30 and not newer_signature(content, closed):
eligible.append(dict(start=begin, end=end, text=content, closed=closed,
month=today.strftime('%Y-%m')))
return eligible

def newer_signature(text, closed):
# English wiki signatures plus ISO datetimes. Revision checks also cover unsigned edits.
months = {m: i for i, m in enumerate(('January', 'February', 'March', 'April', 'May',
'June', 'July', 'August', 'September', 'October', 'November', 'December'), 1)}
for match in re.finditer(r'\b\d{1,2}:\d{2}(?::\d{2})?,?\s+(\d{1,2})\s+('
+ '|'.join(months) + r')\s+(\d{4})\s+\([^)]*\)', text):
try:
d = date(int(match[3]), months[match[2]], int(match[1]))
except ValueError:
return True
if d > closed:
return True
for match in re.finditer(r'\b(\d{4}-\d{2}-\d{2})[T ]\d{2}:\d{2}', text):
try:
if iso(match[1]) > closed:
return True
except ValueError:
return True
return False

def election(text, today):
ts = [t for t in mw.parse(text).filter_templates(recursive=False)
if name(t.name) == 'governance election']
if len(ts) != 1:
return 'needsdate'
try:
p = params(ts[0])
except Halt:
return 'needsdate'
if p.get('status', '').casefold() == 'closed':
return 'closed'
try:
closes = iso(p.get('closes'))
except (ValueError, TypeError):
return 'needsdate'
if p.get('status', '').casefold() != 'open':
return 'needsreview'
if p.get('role', '').casefold() not in ('administrator', 'bureaucrat') or not p.get('candidate'):
return 'needsreview'
return 'overdue' if closes < today else 'notdue'

def nomination_title(title):
return '/' in title and 'preload' not in title.casefold().split('/')[1:]

def validate(c, test=False):
if not isinstance(c, dict) or set(c) != {'version', 'enabled', 'boards', 'report_page'}:
raise Halt('Control schema requires version, enabled, boards, report_page')
if type(c['version']) is not int or c['version'] != 1 or type(c['enabled']) is not bool:
raise Halt('Invalid control version/enabled')
if not isinstance(c['boards'], list) or len(c['boards']) > 7:
raise Halt('At most seven explicit boards')
seen = set()
for b in c['boards']:
if not isinstance(b, dict) or set(b) != {'title', 'archive_prefix'}:
raise Halt('Invalid board schema')
t = b['title']
allowed = isinstance(t, str) and (t in BOARDS if not test else
t.startswith(TEST) and bool(re.fullmatch(r'[A-Za-z0-9 /-]+', t[len(TEST):])))
if not allowed or t in seen or b['archive_prefix'] != t + '/Archive':
raise Halt('Board not allowed, duplicate or incorrect archive prefix')
seen.add(t)
if (not test and c['report_page'] != REPORT) or (test and c['report_page'] != TEST + 'Report'):
raise Halt('Report outside allowed output')
return c

def identity(title, section):
# Content-addressed receipt survives a source-revision change and uncertain saves.
return hashlib.sha256((title + '\0' + section).encode()).hexdigest()

def block(title, section):
key = identity(title, section)
return (f'\n'
f'From [[{title}]], closure explicitly marked by an editor.\n'
+ section + ('' if section.endswith('\n') else '\n')
+ f'')

class Worker:
def __init__(self, site, apply=False, test=False):
self.site, self.apply, self.test = site, apply, test
self.control_title = TEST + 'Control' if test else CONTROL
self.gate = None
self.edits, self.last_write = 0, 0.0

def api(self, **kw):
return self.site.simple_request(maxlag=5, **kw).submit()

def snapshot(self, title):
d = self.api(action='query', titles=title, prop='info|revisions', inprop='protection',
rvslots='main', rvprop='ids|timestamp|content', curtimestamp=1)
q = d['query']
pages = list(q.get('pages', {}).values())
if q.get('interwiki') or len(pages) != 1:
raise Halt('Ambiguous or nonlocal page')
p = pages[0]
if any(k in p for k in ('invalid', 'redirect')):
raise Halt('Redirect or invalid page')
r = p.get('revisions', [{}])[0]
text = r.get('slots', {}).get('main', {}).get('*', '')
if not isinstance(text, str) or len(text.encode()) > MAX_BYTES:
raise Halt('Page exceeds 500 KB bound')
return dict(title=p['title'], text=text, revid=r.get('revid', 0),
timestamp=r.get('timestamp'), start=d['curtimestamp'],
protection=p.get('protection', []), missing='missing' in p,
model=p.get('contentmodel'))

def control(self):
s = self.snapshot(self.control_title)
for kind in ('edit', 'move'):
if not any(p.get('type') == kind and p.get('level') == 'sysop'
and p.get('expiry') in ('infinity', 'infinite', 'indefinite')
for p in s['protection']):
raise Halt('Control requires permanent sysop edit and move protection')
if s['model'] != 'json':
raise Halt('Control requires JSON content model')
try:
c = validate(json.loads(s['text']), self.test)
except (ValueError, TypeError):
raise Halt('Invalid control JSON') from None
if not c['enabled']:
raise Halt('Control disabled')
if self.gate is not None and any(s[k] != self.gate[k] for k in ('text', 'revid', 'protection')):
raise Halt('Control changed during cycle')
self.gate = s
return c

def save(self, old, text, summary, prerequisite=None):
# MediaWiki strips trailing page whitespace during its pre-save transform.
# Canonicalise before comparison; receipt-delimited section bytes stay intact.
text = text.rstrip()
if not self.apply:
raise Halt('Write attempted during dry-run')
if text == old['text']:
return old
if self.edits >= 12 or len(text.encode()) > MAX_BYTES:
raise Halt('Edit or page-size bound reached')
optout(old['text'])
# Pace before the final observations, never sleep with a stale guard.
time.sleep(max(0, 15 - (time.monotonic() - self.last_write)))
token = self.site.tokens['csrf']
self.control()
fresh = self.snapshot(old['title'])
if any(fresh[k] != old[k] for k in ('text', 'revid', 'protection', 'missing')):
raise Halt('Target changed before save')
optout(fresh['text'])
# Cross-page prerequisite must run AFTER pacing and final source observation.
# The edit request immediately follows; never sleep after this check.
if prerequisite is not None:
prerequisite()
kw = dict(action='edit', title=old['title'], text=text, summary=summary,
token=token, bot=1, assertuser=BOT, starttimestamp=old['start'])
if old['missing']:
kw['createonly'] = 1
else:
# Use revid alone: basetimestamp permits self-conflicts on some MW versions.
kw.update(nocreate=1, baserevid=old['revid'])
self.edits += 1
self.last_write = time.monotonic()
try:
result = self.api(**kw)['edit']
except Exception:
raise Halt('Save conflict or uncertain result; inspect pages before retry; no further edits') from None
if result.get('result') != 'Success':
raise Halt('Save not confirmed')
check = self.snapshot(old['title'])
if check['text'] != text or check['revid'] != result.get('newrevid'):
raise Halt('Save verification failed; no further edits')
return check

def archive_one(self, title, prefix, today):
src = self.snapshot(title)
if src['missing']:
return 'missing'
optout(src['text'])
eligible = sections(src['text'], today)
if not eligible:
return 'unchanged'
section = eligible[0]
if not self.quiet_since(src, section):
return 'recent-or-unverifiable-activity'
# Identical repeated sections cannot be safely distinguished by content receipts.
if src['text'].count(section['text']) != 1:
raise Halt('Ambiguous duplicate source section')
target = self.intent(src, section, src['title'] + '/Archive ' + section['month'])
dest = self.snapshot(target)
optout(dest['text'])
payload = block(src['title'], section['text'])
receipt = 'EliteGovernance:archive:' + identity(src['title'], section['text'])
if receipt in dest['text'] and (dest['text'].count(payload) != 1
or dest['text'].count(receipt) != 2):
raise Halt('Archive receipt exists but differs; source retained')
if not self.apply:
return 'would-archive'
if payload not in dest['text']:
# Observe source again before making an archive copy.
current = self.snapshot(title)
if current['revid'] != src['revid'] or current['text'] != src['text']:
raise Halt('Source changed before archive; retained')
dest = self.save(dest, dest['text'] + ('\n' if dest['text'] else '') + payload,
'Archive from [[' + src['title'] + ']] oldid=' + str(src['revid'])
+ '; author/timestamp attribution in source history')
# Critical ordering: archive content is reread and verified before source removal.
verified = self.snapshot(dest['title'])
optout(verified['text'])
if verified['text'].count(payload) != 1:
raise Halt('Archive copy not verified; source retained')
def still_archived():
final = self.snapshot(dest['title'])
optout(final['text'])
if (final['missing'] or final['revid'] != verified['revid']
or final['text'] != verified['text'] or final['text'].count(payload) != 1):
raise Halt('Archive changed before source removal; source retained')
self.save(src, src['text'][:section['start']] + src['text'][section['end']:],
'Move verified closed discussion to [[' + dest['title'] + ']]',
prerequisite=still_archived)
self.intent(src, section, target, complete=True)
return 'archived'

def intent(self, src, section, target, complete=False):
"""Persist before archive write; retries across month boundaries use that copy.

Separate test journal, <=7 pending boards. If content changes after a partial
transfer, retain both copies and require human inspection rather than guessing.
"""
path = Path(os.environ.get('GOVERNANCE_STATE_DIR', '/var/lib/elitewiki-governance')) / (
'test-intents.json' if self.test else 'intents.json')
try:
data = json.loads(path.read_text())['intents'] if path.exists() else {}
if not isinstance(data, dict) or len(data) > 7:
raise ValueError()
except (ValueError, KeyError, TypeError):
raise Halt('Invalid durable intents; inspect state before retry') from None
key, digest = src['title'], identity(src['title'], section['text'])
previous = data.get(key)
if previous:
if not re.fullmatch(re.escape(src['title'] + '/Archive ') + r'\d{4}-\d{2}', previous.get('target', '')):
raise Halt('Pending archive content differs; human inspection required')
if previous.get('hash') != digest:
# Recover a crash after successful source removal but before clearing
# the intent. Never clear an intent while its original section remains.
prior_text = previous.get('section', '')
if (not prior_text or prior_text in src['text']
or identity(key, prior_text) != previous.get('hash')):
raise Halt('Pending archive content differs; human inspection required')
archived = self.snapshot(previous['target'])
optout(archived['text'])
if archived['text'].count(block(key, prior_text)) != 1:
raise Halt('Pending archive not verified; human inspection required')
data.pop(key)
previous = None
else:
target = previous['target']
if self.apply:
if complete:
data.pop(key, None)
elif not previous:
if len(data) >= 7:
raise Halt('Pending archive bound reached; inspect state')
data[key] = dict(hash=digest, target=target, source_revid=src['revid'], section=section['text'])
else:
return target
status(path, dict(intents=data))
return target

def quiet_since(self, src, section):
"""Prove identical complete section since end of closure date, fail closed.

Up to 20 newer board revisions plus one boundary revision. No continuation,
full-history scans or stored history database. Busy boards defer to humans.
"""
boundary = section['closed'].isoformat() + 'T23:59:59Z'
if src['timestamp'] <= boundary:
return True
common = dict(action='query', titles=src['title'], prop='revisions', rvslots='main',
rvprop='ids|timestamp|content')
older = self.api(**common, rvstart=boundary, rvlimit=1)
recent = self.api(**common, rvend=boundary, rvlimit=20)
if 'continue' in recent:
return False
pages = list(older['query']['pages'].values()) + list(recent['query']['pages'].values())
if any(not p.get('revisions') for p in pages):
return False
revisions = [r for p in pages for r in p['revisions']]
for rev in revisions:
text = rev.get('slots', {}).get('main', {}).get('*')
if not isinstance(text, str) or len(text.encode()) > MAX_BYTES:
return False
# Exact parsed section boundary, not a matching substring in another section.
matches = [s for s in sections(text, section['closed'] + timedelta(days=31))
if s['text'] == section['text']]
if len(matches) != 1:
return False
return True

def discover(self):
titles, truncated = [], False
# Two fixed categories, at most 100 unique election reads, no category recursion.
for category in CATEGORIES:
d = self.api(action='query', list='categorymembers', cmtitle=category,
cmtype='page', cmnamespace=4, cmlimit=100, cmprop='title')
truncated |= 'continue' in d
for p in d['query']['categorymembers']:
t = p['title']
if not nomination_title(t):
continue
if t not in titles:
if len(titles) == 100:
truncated = True
else:
titles.append(t)
return titles, truncated

def report(self, title, titles, truncated, today):
rows = []
for t in titles:
if not nomination_title(t):
continue
# Election pages must be subpages, even when categories contain bad entries.
if '/' not in t or (self.test and not t.startswith(TEST)):
rows.append((t, 'needsreview'))
continue
s = self.snapshot(t)
state = 'needsdate' if s['missing'] else election(s['text'], today)
if state not in ('closed', 'notdue'):
rows.append((t, state))
# Human text is outside a single managed region. No hourly timestamp churn.
lines = ['Election deadline checks only. No vote counts, closures or results.',
'Operator: [[User:Talondri|Talondri]] (site operator). '
'[[User:EliteMaintenanceBot/Governance source|Public source and scope]].']
if truncated:
lines.append('Discovery limit reached: at most 100 pages checked; human review needed.')
lines += [f'* [[{t}]]: {state}' for t, state in sorted(rows)]
if not rows:
lines.append('No overdue or undated elections in the bounded category check.')
body = '\n' + '\n'.join(lines) + '\n'
s = self.snapshot(title)
optout(s['text'])
if s['missing']:
text = '{{bots|allow=EliteMaintenanceBot}}\n' + START + body + END
else:
if s['text'].count(START) != 1 or s['text'].count(END) != 1:
raise Halt('Report needs exactly one managed marker pair')
a, b = s['text'].index(START) + len(START), s['text'].index(END)
if a > b:
raise Halt('Report markers reversed')
text = s['text'][:a] + body + s['text'][b:]
if self.apply and text != s['text']:
self.save(s, text, 'Report election deadlines for human review only')
return dict(findings=rows, discovery_truncated=truncated, checked=len(titles))

def run(self, today):
c = self.control()
results = []
# One section per board per cycle, <=7 archive pairs, total hard cap 12 edits.
# Reserve one edit for the report; remaining backlog waits for next hour.
for b in c['boards']:
if self.edits > 9:
break
try:
state = self.archive_one(b['title'], b['archive_prefix'], today)
except Halt as e:
if str(e) == 'Bot exclusion':
state = 'excluded'
else:
raise
results.append(dict(title=b['title'], status=state))
titles, truncated = ([], False) if self.test else self.discover()
report = self.report(c['report_page'], titles, truncated, today)
return dict(status='applied' if self.apply else 'dry-run', boards=results,
elections=report, edits=self.edits)

@contextmanager
def lock(path):
import fcntl
with open(path, 'a') as f:
try:
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
raise Halt('Governance worker already active') from None
yield

def status(path, result):
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix='.status-')
try:
with os.fdopen(fd, 'w') as stream:
json.dump(dict(result, at=datetime.now(timezone.utc).isoformat()), stream, indent=2)
stream.flush()
os.fsync(stream.fileno())
os.replace(tmp, path)
finally:
if os.path.exists(tmp):
os.unlink(tmp)

def connect():
import pywikibot
pywikibot.config.max_retries = 1
pywikibot.config.retry_wait = 5
pywikibot.config.retry_max = 10
pywikibot.config.socket_timeout = (10, 30)
pywikibot.config.log = []
site = pywikibot.Site()
if site.hostname() != 'elite.mistflux.co.uk' or site.protocol() != 'https':
raise Halt('Unexpected wiki endpoint')
site.login()
if site.user() != BOT or 'bot' not in site.userinfo.get('rights', []):
raise Halt('Expected EliteMaintenanceBot with bot right')
return site

def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--apply', action='store_true')
p.add_argument('--test', action='store_true', help='Only isolated user-subpage control and outputs')
args = p.parse_args()
directory = Path(os.environ.get('GOVERNANCE_STATE_DIR', '/var/lib/elitewiki-governance'))
try:
with lock(directory / 'governance.lock'):
try:
result = Worker(connect(), args.apply, args.test).run(datetime.now(timezone.utc).date())
code = 0
except Halt as e:
result, code = dict(status='halted', reason=str(e)), 2
except Exception as e:
result, code = dict(status='error', error_type=type(e).__name__), 1
status(directory / ('test-status.json' if args.test else 'status.json'), result)
print(json.dumps(result), flush=True)
return code
except Halt as e:
print(str(e))
return 2

if __name__ == '__main__':
sys.exit(main())

</syntaxhighlight>

User:EliteMaintenanceBot/Governance source - Revision history

Talondri: Protected "User:EliteMaintenanceBot/Governance source": Preserve reviewed operational bot source ([Edit=Allow only administrators] (indefinite) [Move=Allow only administrators] (indefinite))

Talondri: Publish reviewed secret-free governance worker and authorised task scope