Add deterministic monthly PQRST report and disabled schedule projection

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e759-301a-78b1-bbc1-040ef094b12d

src/activity_core/context_resolvers/__init__.py

@@ -9,3 +9,5 @@ from activity_core.context_resolvers import (  # noqa: F401
     state_hub,
     reuse_surface,
 )
+
+from activity_core.context_resolvers import pqrst  # noqa: F401

src/activity_core/context_resolvers/pqrst.py

@@ -0,0 +1,231 @@
+"""Monthly, read-only PQRST statistics from a pinned public hall Git tree.
+
+One selected published block per agent entry; never infer first-attempt success,
+weight by time/tokens, execute entry text, or compare workers. Missing observations
+are null, not zero. Collection fails closed rather than reporting partial means.
+"""
+from __future__ import annotations
+
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+from datetime import date, datetime, timezone
+import hashlib
+import json
+import re
+import time
+from urllib.parse import quote
+from zoneinfo import ZoneInfo
+
+import httpx
+import yaml
+
+from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
+
+HOST = 'https://forgejo.coulomb.social'
+REPO = 'coulomb/hall-of-helix'
+TZ = ZoneInfo('Europe/Berlin')
+LETTERS = 'PQRST'
+SHA = re.compile(r'[a-f0-9]{40}')
+SIG = re.compile(r'P(\d{1,3}) Q(\d{1,3}) R(\d{1,3}) S(\d{1,3}) T(\d{1,3})')
+MAX_ENTRIES = 5000
+
+
+class StrictLoader(yaml.SafeLoader):
+    pass
+
+
+def mapping(loader, node, deep=False):
+    result = {}
+    for key_node, value_node in node.value:
+        key = loader.construct_object(key_node, deep=deep)
+        if key in result:
+            raise ValueError('duplicate_metadata')
+        result[key] = loader.construct_object(value_node, deep=deep)
+    return result
+
+
+StrictLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, mapping)
+
+
+def signature(value):
+    match = SIG.fullmatch(value) if isinstance(value, str) else None
+    return tuple(map(int, match.groups())) if match else None
+
+
+def parse_entry(text):
+    """Return metadata, date, attempt count and selected structural result."""
+    fm = re.match(r'\A---\s*\n(.*?)\n---\s*\n', text, re.S)
+    if not fm:
+        raise ValueError('metadata')
+    meta = yaml.load(fm[1], Loader=StrictLoader)
+    if not isinstance(meta, dict):
+        raise ValueError('metadata')
+    stamp = meta.get('recorded_at')
+    if isinstance(stamp, datetime):
+        if stamp.tzinfo is None:
+            raise ValueError('date')
+        day = stamp.astimezone(TZ).date()
+    elif isinstance(stamp, date):
+        day = stamp
+    elif isinstance(stamp, str) and re.fullmatch(r'\d{4}-\d{2}-\d{2}', stamp):
+        day = date.fromisoformat(stamp)
+    elif isinstance(stamp, str):
+        parsed = datetime.fromisoformat(stamp.replace('Z', '+00:00'))
+        if parsed.tzinfo is None:
+            raise ValueError('date')
+        day = parsed.astimezone(TZ).date()
+    else:
+        raise ValueError('date')
+    sections = re.findall(r'^## PQRST estimate[^\S\n]*\n(.*?)(?=^## |\Z)', text[fm.end():], re.M | re.S | re.I)
+    blocks = [m[1] for section in sections for m in re.finditer(
+        r'^```[^\n]*\n(PQRST-Estimate\n.*?)^```[^\S\n]*$', section, re.M | re.S)]
+    if not sections and not meta.get('pqrst_estimate'):
+        return meta, day, len(blocks), None, 'missing'
+    target = signature(meta.get('pqrst_estimate'))
+    selected = []
+    for block in blocks:
+        fields = {}
+        for field in re.finditer(r'^([A-Za-z][A-Za-z ]*):[^\S\n]*(.*?)(?=^[A-Za-z][A-Za-z ]*:|\Z)', block, re.M | re.S):
+            fields.setdefault(field[1], []).append(field[2].strip())
+        sigs = fields.get('Signature', [])
+        if len(sigs) == 1 and target is not None and signature(sigs[0]) == target:
+            selected.append(fields)
+    if len(sections) != 1 or len(selected) != 1:
+        return meta, day, len(blocks), None, 'invalid'
+    fields = selected[0]
+    required = [*LETTERS, 'Sum', 'Confidence', 'Signature', 'Dominant factors']
+    if any(len(fields.get(k, [])) != 1 for k in required) or any(k not in required + ['Notes'] for k in fields):
+        return meta, day, len(blocks), None, 'invalid'
+    values = []
+    for key in LETTERS:
+        match = re.fullmatch(r'(\d{1,3})%', fields[key][0])
+        if not match:
+            return meta, day, len(blocks), None, 'invalid'
+        values.append(int(match[1]))
+    confidence = fields['Confidence'][0]
+    if (sum(values) != 100 or any(v > 100 for v in values) or tuple(values) != target
+            or fields['Sum'] != ['100%'] or confidence not in {'low', 'medium', 'high'}
+            or not fields['Dominant factors'][0]):
+        return meta, day, len(blocks), None, 'invalid'
+    return meta, day, len(blocks), (values, confidence), 'included'
+
+
+def month_keys(now):
+    current = now.astimezone(TZ)
+    number = current.year * 12 + current.month - 1
+    return [f'{(number-i)//12:04d}-{(number-i)%12+1:02d}' for i in range(6, 0, -1)]
+
+
+def summarize(entries, revision, now=None):
+    now = now or datetime.now(timezone.utc)
+    if now.tzinfo is None:
+        raise ValueError('timezone_required')
+    months = month_keys(now)
+    counts = Counter(entries=len(entries))
+    buckets = {key: {'entries': 0, 'missing': 0, 'invalid': 0, 'attempts': 0,
+                     'drafts': 0, 'values': [], 'confidence': Counter()} for key in months}
+    parsed = []
+    for path, text in entries:
+        try:
+            result = parse_entry(text)
+            parsed.append(result)
+        except (ValueError, TypeError, KeyError, OverflowError, yaml.YAMLError):
+            counts['unclassifiable'] += 1
+    identities = Counter(m.get('id') for m, *_ in parsed if isinstance(m.get('id'), str))
+    for meta, day, attempts, result, disposition in parsed:
+        if meta.get('worker_kind') != 'agent-session':
+            counts['non_agent'] += 1
+            continue
+        key = day.strftime('%Y-%m')
+        if key not in buckets:
+            counts['outside_window'] += 1
+            continue
+        bucket = buckets[key]
+        bucket['entries'] += 1
+        bucket['attempts'] += attempts
+        if not isinstance(meta.get('id'), str) or identities[meta['id']] != 1:
+            disposition = 'invalid'
+        if disposition != 'included':
+            bucket[disposition] += 1
+            continue
+        values, confidence = result
+        bucket['values'].append(values)
+        bucket['confidence'][confidence] += 1
+        bucket['drafts'] += meta.get('status') == 'draft'
+    series = []
+    previous = None
+    for month, bucket in buckets.items():
+        values = bucket.pop('values')
+        n = len(values)
+        means = [sum(v[i] for v in values) / n for i in range(5)] if n else None
+        delta = [round(a-b, 2) for a, b in zip(means, previous)] if means is not None and previous is not None else None
+        series.append(dict(month=month, n=n, mean_pp=[round(v, 2) for v in means] if means else None,
+                           change_pp=delta, **bucket))
+        previous = means
+    latest = series[-1]
+    return {'repo': 'pqrst-practice', 'source': f'{REPO}@{revision}',
+            'generated_at': now.isoformat(), 'timezone': 'Europe/Berlin',
+            'status': 'needs_review' if counts['unclassifiable'] else 'ok',
+            'summary': f"PQRST {latest['month']}: {latest['n']} structurally valid published datapoints; equal-entry monthly means and month-over-month changes in P/Q/R/S/T percentage points.",
+            'order': list(LETTERS), 'counts': dict(counts), 'months': series,
+            'limits': 'Published agent entries, including drafts; not first attempts or a census. Structural checks cannot certify evidence quality. No worker rankings, causal inference or significance claim; compare sample counts and task mix. Empty means/deltas are null. Rounding may shift totals by 0.01.',
+            'model_calls': 0}
+
+
+def collect(now=None):
+    """Fixed-host, commit-pinned, bounded collection. Never return partial means."""
+    deadline = time.monotonic() + 120
+    def get(client, url, limit):
+        if time.monotonic() >= deadline:
+            raise ValueError('collection_deadline')
+        with client.stream('GET', url) as response:
+            response.raise_for_status()
+            body = bytearray()
+            for part in response.iter_bytes():
+                body.extend(part)
+                if len(body) > limit or time.monotonic() >= deadline:
+                    raise ValueError('collection_limit')
+            return bytes(body)
+    with httpx.Client(timeout=10, follow_redirects=False) as client:
+        info = json.loads(get(client, f'{HOST}/api/v1/repos/{REPO}/branches/main', 65536))
+        sha = info['commit']['id']
+        if not isinstance(sha, str) or not SHA.fullmatch(sha):
+            raise ValueError('source_revision')
+        tree = json.loads(get(client, f'{HOST}/api/v1/repos/{REPO}/git/trees/{sha}?recursive=true&per_page=10000', 4_000_000))
+        if tree.get('truncated') is not False or len(tree['tree']) != tree['total_count']:
+            raise ValueError('incomplete_tree')
+        files = [f for f in tree['tree'] if f['path'].startswith('entries/') and f['path'].endswith('.md')]
+        if len(files) > MAX_ENTRIES or not files:
+            raise ValueError('entry_count')
+        def read(f):
+            path = f['path']
+            if (f.get('mode') != '100644' or f.get('type') != 'blob'
+                    or '..' in path.split('/') or not SHA.fullmatch(f.get('sha', ''))):
+                raise ValueError('entry_path')
+            data = get(client, f'{HOST}/{REPO}/raw/commit/{sha}/{quote(path, safe="/")}', 131072)
+            blob = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
+            if blob != f['sha']:
+                raise ValueError('entry_hash')
+            return path, data.decode('utf-8')
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            entries = list(pool.map(read, files))
+    return summarize(entries, sha, now)
+
+
+class PqrstResolver(ContextResolver):
+    def resolve(self, query, event, params):
+        if query != 'monthly':
+            raise ValueError('unsupported_pqrst_query')
+        try:
+            report = collect()
+        except (ValueError, KeyError, TypeError, UnicodeError, httpx.HTTPError):
+            report = {'repo': 'pqrst-practice', 'status': 'unavailable',
+                      'summary': 'PQRST source collection failed; no partial averages published.',
+                      'months': [], 'model_calls': 0}
+        encoded = json.dumps(report, sort_keys=True, separators=(',', ':'))
+        if len(encoded) > 3900:
+            raise ValueError('pqrst_digest_limit')
+        return encoded
+
+
+CONTEXT_RESOLVER_REGISTRY['pqrst'] = PqrstResolver