troupe-standup/wizard-work/ledger/check_workqueue_qcensus.py

import hashlib
import pathlib
import re
import sys

SRC = '/root/troupe-standup/WORK-QUEUE.md'
TSV = '/root/troupe-standup/wizard-work/ledger/workqueue-q-census.tsv'

def _strip_decorations(s):
    s = s.rstrip()
    start = s.rfind('*(')
    if start == -1:
        return s
    i = start + 1
    depth = 0
    for j in range(i, len(s)):
        if s[j] == '(':
            depth += 1
        elif s[j] == ')':
            depth -= 1
            if depth == 0:
                return s[:start] + s[j + 2:].strip()
    return s

def parse_sections(text):
    lines = text.splitlines()
    sections = []
    counts = {}
    cur = None
    for raw in lines:
        m = re.match(r'^## (Q\d+)(.*)$', raw)
        if m:
            if cur is not None:
                sections.append(cur)
            qid = m.group(1)
            rest = m.group(2).strip()
            is_closure = rest.startswith('closure')
            if not is_closure:
                counts[qid] = counts.get(qid, 0) + 1
                qid_out = qid if counts[qid] == 1 else f'{qid} #{counts[qid]}'
            else:
                qid_out = qid + ' closure'
            title = clean_title(_strip_decorations(rest))
            cur = {
                'q_id': qid_out,
                'rest': rest,
                'title': title,
                'is_closure': is_closure,
                'status_raw': None,
                'body_lines': 0,
            }
            continue
        if cur is not None:
            cur['body_lines'] += 1
            sm = re.match(r'\*\*Status:\s*(.*)$', raw)
            if sm and cur['status_raw'] is None:
                s = sm.group(1).strip()
                if s.endswith('**'):
                    s = s[:-2].rstrip()
                cur['status_raw'] = s
    if cur is not None:
        sections.append(cur)
    return sections

def clean_title(title):
    return title.lstrip('—-').strip()

def main():
    text = pathlib.Path(SRC).read_text()
    sections = parse_sections(text)

    errors = []

    # 1. Section count
    if len(sections) != 97:
        errors.append(f'expected 97 sections, got {len(sections)}')

    # 2. Closure count
    closures = [s for s in sections if s['is_closure']]
    if len(closures) != 8:
        errors.append(f'expected 8 closures, got {len(closures)}')

    # 3. Deterministic regeneration: recompute TSV content and diff md5 vs committed
    rows = []
    for s in sections:
        rows.append('\t'.join([
            s['q_id'], s['rest'].rstrip(), s['status_raw'] or '',
            re.search(r'[a-f0-9]{7,}', s['rest'] + ' ' + s['q_id']).group(0) if re.search(r'[a-f0-9]{7,}', s['rest'] + ' ' + s['q_id']) else '',
            str(s['body_lines']),
            '1' if s['is_closure'] else '0',
        ]))
    expected_tsv = '#q_id\tannex\tstatus_raw\tkey_hash\tbody_lines\tclosure\n'
    for r in rows:
        expected_tsv += r + '\n'
    expected_md5 = hashlib.md5(expected_tsv.encode('utf-8')).hexdigest()
    actual_md5 = hashlib.md5(pathlib.Path(TSV).read_bytes()).hexdigest()
    if expected_md5 != actual_md5:
        errors.append(f'md5 mismatch: expected {expected_md5}, got {actual_md5}')

    # 4. Anchor checks via a lookup map
    lookup = {}
    for s in sections:
        lookup[s['q_id']] = s

    def norm(s):
        if s is None:
            return ''
        return s

    def check_attr(qid, attr, expected_contains):
        if qid not in lookup:
            errors.append(f'{qid} missing')
            return
        v = norm(lookup[qid].get(attr))
        if expected_contains not in v:
            errors.append(f'{qid}.{attr} mismatch: {v!r} does not contain {expected_contains!r}')

    # 5. Exact title anchor
    if 'Q1' not in lookup or lookup['Q1']['rest'].lstrip('—-').strip() != 'Recover the fifteen silent rooms':
        errors.append('Q1 title mismatch')

    def status_of(qid):
        return norm(lookup.get(qid, {}).get('status_raw'))

    def hash_of(qid):
        h = re.search(r'[a-f0-9]{7,}', lookup.get(qid, {}).get('rest', '') + ' ' + qid)
        return h.group(0) if h else ''

    if 'CLOSED — 0/15 recovered' not in norm(status_of('Q1')):
        errors.append('Q1 status mismatch')
    if 'DONE — recovered.' not in norm(status_of('Q2')):
        errors.append('Q2 status mismatch')
    if 'OPEN — awaiting Alexander' not in norm(status_of('Q45')):
        errors.append('Q45 status mismatch')
    if 'open and unclaimed' not in norm(status_of('Q75')):
        errors.append('Q75 status mismatch')
    if 'open and unclaimed' not in norm(status_of('Q88')):
        errors.append('Q88 status mismatch')
    if 'DONE at `9878f12`' not in norm(status_of('Q93 closure')):
        errors.append('Q93 closure status mismatch')
    if 'DONE' not in norm(status_of('Q15')):
        errors.append('Q15 status mismatch')

    if hash_of('Q75') != 'b82b023':
        errors.append(f'Q75 hash {hash_of("Q75")!r}')
    if hash_of('Q88') != 'b0a2886':
        errors.append(f'Q88 hash {hash_of("Q88")!r}')
    if hash_of('Q93 closure') != 'd56105a':
        errors.append(f'Q93 closure hash {hash_of("Q93 closure")!r}')

    # 6. Uniqueness of q_ids
    qids = [s['q_id'] for s in sections]
    if len(set(qids)) != len(qids):
        errors.append('duplicate q_ids present')

    if errors:
        print('FAIL')
        for e in errors:
            print(' -', e)
        sys.exit(1)
    print('PASS: 97 sections, 8 closures, regeneration md5 match, all anchors verified')
    sys.exit(0)

if __name__ == '__main__':
    main()