diff --git a/REPORTING.md b/REPORTING.md index 11c479b..45012d7 100644 --- a/REPORTING.md +++ b/REPORTING.md @@ -135,3 +135,35 @@ two-minute collection time budget. Therefore “every five minutes” is the col attempt interval, not a guarantee that every repository is current. Event time, source check time and page check time remain distinct. Routine feed rebuilds never create activity events. + +### Pulse coverage and fair collection + +`config/activity_sources.json` routes the registered `autolens_profiling` and +`autofit_profiling` development repositories to Pulse. Their identity still comes +from Mind's body map. A repository gets at most 12 API requests in a cycle; +partial normalized evidence survives to the next cycle and other repositories +get a turn. The whole-cycle request and two-minute limits still apply. GitHub +schedule delivery remains best effort; an enabled schedule is not evidence that +a collection occurred. The published source check times are the evidence. + +The Pulse adapter reads `campaigns.yaml`, `registry.yaml` and the registered +`snapshots/.json` at one verified public Pulse commit. Campaign and task +changes share one campaign row. The recorded `last_checkin` dates a check-in, +not a measurement. Repeated check-ins with unchanged facts and next steps do not +create new events. `active` is not displayed as `running`. + +A measurement event requires a producer `provenance.measured_at`, a stable +`run_id` and an explicit campaign association. Pulse may provide the association +as `campaigns[].runs: [{repo: , run_id: }]` in its +own ledger; alternatively a producer may declare `campaign_id` where its schema +allows it. The adapter never guesses from setup names or filenames. Multiple +measurements from a run produce one campaign-history entry; qualification, +speedups, comparisons and scientific acceptance are not inferred. No run-start +event is invented from a later measurement. Current legacy records lacking times +are counted in coverage and excluded from the timeline. + +No Pulse or project records are changed by collection. Pulse input errors keep +last-known campaign history marked stale; an unverified public Pulse repository +suppresses it. The new adapter does not run profiling or add project collection +approval. The original development-only scope above describes PR84; this section +extends it with recorded Pulse campaign and admissible run activity. diff --git a/config/activity_sources.json b/config/activity_sources.json new file mode 100644 index 0000000..a004e23 --- /dev/null +++ b/config/activity_sources.json @@ -0,0 +1,8 @@ +{ + "schema_version": 1, + "board_routes": { + "autolens_profiling": "pulse", + "autofit_profiling": "pulse" + }, + "pulse": {"repository": "PyAutoPulse", "ref": "main"} +} diff --git a/scripts/activity_feed.py b/scripts/activity_feed.py index 45f3f5b..d6a54bf 100644 --- a/scripts/activity_feed.py +++ b/scripts/activity_feed.py @@ -43,7 +43,7 @@ def short(text, limit=110): return re.sub(r'\s+', ' ', str(text or '')).strip()[:limit] -def sources(path): +def sources(path, routes=None): import yaml rows = yaml.safe_load(Path(path).read_text())['repos'] result = {} @@ -52,7 +52,9 @@ def sources(path): if not re.fullmatch(r'[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+', repo): continue organ = str(row.get('organ', 'scientist' if name == 'PyAutoScientist' else 'mind')).lower() - board = organ if organ in BOARDS else 'mind' + board = (routes or {}).get(name, organ if organ in BOARDS else 'mind') + if board not in BOARDS: + raise ValueError('unknown activity board') result[repo] = {'board': board, 'board_url': f'https://{repo.split("/")[0].lower()}.github.io/PyAuto{BOARDS[board]}/'} return dict(sorted(result.items())) @@ -104,6 +106,20 @@ def pages(self, path, params): raise error +class RepositoryBudget: + """A busy repo yields to peers; normalized partial batches resume next cycle.""" + def __init__(self, parent, limit=12): + self.parent, self.start, self.limit = parent, parent.requests, limit + + def get(self, path): + if self.parent.requests - self.start >= self.limit: + raise CollectionError('repository allowance reached') + return self.parent.get(path) + + def pages(self, path, params): + return GitHub.pages(self, path, params) + + def closing_issue(body): """Only one explicit, same-repository closing directive; no fuzzy pairing.""" body = re.sub(r'```.*?```', '', body or '', flags=re.S) @@ -207,7 +223,7 @@ def page_set(path, params): return result -def collect(registry, previous, api, now=None): +def collect(registry, previous, api, now=None, pulse_config=None, identities=None): now = now or datetime.now(timezone.utc) previous = previous if previous.get('schema_version') == SCHEMA else {} cache = {'schema_version': SCHEMA, 'sources': {}, 'next': 0} @@ -234,6 +250,11 @@ def collect(registry, previous, api, now=None): cache['sources'][repo] = dict(old, status='stale', reason='awaiting collection') else: cache['sources'][repo] = {'status': 'unavailable', 'reason': 'public source unavailable', 'events': {}, 'tasks': {}} + if pulse_config: + from pulse_activity import collect_pulse + cache['pulse'] = collect_pulse(RepositoryBudget(api), (identities or {}).get(pulse_config['repository'], ''), + pulse_config.get('ref', 'main'), public, identities or {}, + previous.get('pulse', {}), now) advanced = 0 for repo in ordered: if repo not in public: @@ -242,7 +263,7 @@ def collect(registry, previous, api, now=None): old = cache['sources'][repo] since = stamp(max(now - timedelta(days=RETENTION_DAYS), instant(old['cursor']) - timedelta(minutes=2))) if old.get('cursor') else stamp(now - timedelta(days=7)) try: - cache['sources'][repo] = collect_repo(api, repo, since, now, old) + cache['sources'][repo] = collect_repo(RepositoryBudget(api) if isinstance(api, GitHub) else api, repo, since, now, old) advanced += 1 except (CollectionError, ValueError, KeyError, TypeError) as exc: old['status'] = 'stale' @@ -280,8 +301,13 @@ def make_feed(registry, cache, now): row['history_count'] = len(row['events']) row['events'] = row['events'][:20] row['time'] = row['events'][0]['time'] + if 'pulse' in cache: + from pulse_activity import publish + coverage.append({'repo': 'Pulse campaign evidence', **{k: cache['pulse'].get(k) for k in ('status', 'checked', 'reason')}}) + for row in publish(cache['pulse'], now): + grouped[row['id']] = row rows = sorted(grouped.values(), key=lambda row: (row['time'], row['id']), reverse=True) - return {'schema_version': SCHEMA, 'generated': stamp(now), 'scope': 'Recorded development activity', + return {'schema_version': SCHEMA, 'generated': stamp(now), 'scope': 'Development and Pulse campaign activity', 'retention_days': RETENTION_DAYS, 'initial_backfill_days': 7, 'truncated': len(rows) > MAX_TASKS, 'coverage': coverage, 'tasks': rows[:MAX_TASKS]} @@ -320,11 +346,11 @@ def render_html(feed=None): data = json.dumps(feed, ensure_ascii=False).replace('<', '\\u003c').replace('>', '\\u003e').replace('&', '\\u0026') return f'''

Live Feed

-

Recorded development activity

Collection {esc(feed.get('generated') or 'not yet available')}

+

Development and Pulse campaign activity

Collection {esc(feed.get('generated') or 'not yet available')}

    {''.join(rows)}

No recorded activity available yet.

-
Coverage and freshness

Public recorded development only; scientific runs and unrecorded local work are not included. Up to 30 days retained; initial collection looks back seven days. Collection runs approximately every five minutes; repositories share a bounded request budget.

    +
    Coverage and freshness

    Public development and Pulse campaign check-ins; only explicitly associated, dated run evidence enters the timeline. Unrecorded local work is not included. Up to 30 days retained; initial collection looks back seven days. Collection runs approximately every five minutes; repositories share a bounded request budget.

      ''' @@ -335,10 +361,18 @@ def main(): ap.add_argument('--cache', required=True, type=Path) ap.add_argument('--output', required=True, type=Path) ap.add_argument('--budget', type=int, default=60) + ap.add_argument('--config', type=Path, default=Path(__file__).resolve().parents[1] / 'config' / 'activity_sources.json') args = ap.parse_args() # 180 calls / 5min fits a 5000/hour PAT. Default GITHUB_TOKEN uses 60/cycle. previous = json.loads(args.cache.read_text()) if args.cache.exists() else {} - cache, feed = collect(sources(args.repos), previous, GitHub(os.environ.get('GH_TOKEN'), args.budget)) + import yaml + config = json.loads(args.config.read_text()) + if config.get('schema_version') != 1: + raise ValueError('unsupported activity source config') + identities = {name: row.get('github') for name, row in yaml.safe_load(args.repos.read_text())['repos'].items()} + cache, feed = collect(sources(args.repos, config.get('board_routes')), previous, + GitHub(os.environ.get('GH_TOKEN'), args.budget), + pulse_config=config.get('pulse'), identities=identities) for path, value in ((args.cache, cache), (args.output, feed)): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n') @@ -346,4 +380,6 @@ def main(): if __name__ == '__main__': + import sys + sys.modules['activity_feed'] = sys.modules[__name__] main() diff --git a/scripts/live_feed.js b/scripts/live_feed.js index f5fbddf..89f3c69 100644 --- a/scripts/live_feed.js +++ b/scripts/live_feed.js @@ -57,7 +57,7 @@ function renderCoverage() { const gaps = current.coverage.filter(c => c.status !== 'ok').length; const aged = current.generated && Date.now() - Date.parse(current.generated) > 15 * 60000; fresh.textContent = current.generated ? `Collected ${age(current.generated)}${aged ? ' · delayed' : ''}${lastChecked ? ' · checked ' + age(lastChecked) : ''}` : 'Collection not yet available'; - root.querySelector('[data-feed-coverage-label]').textContent = `Coverage and freshness · ${current.coverage.length - gaps}/${current.coverage.length} sources checked${gaps ? ' · gaps' : ''}${current.truncated ? ' · latest 200 tasks' : ''}`; + root.querySelector('[data-feed-coverage-label]').textContent = `Coverage and freshness · ${current.coverage.length - gaps}/${current.coverage.length} sources complete${gaps ? ' · gaps' : ''}${current.truncated ? ' · latest 200 tasks' : ''}`; coverage.replaceChildren(...current.coverage.map(c => el('li', `${c.repo}: ${c.status}${c.checked ? ' · checked ' + age(c.checked) : ''}${c.reason ? ' · ' + c.reason : ''}`))); } function reading() { diff --git a/scripts/pulse_activity.py b/scripts/pulse_activity.py new file mode 100644 index 0000000..1e8ce47 --- /dev/null +++ b/scripts/pulse_activity.py @@ -0,0 +1,163 @@ +"""Read Pulse-owned, public campaign evidence without executing or judging it.""" +from __future__ import annotations +import base64 +from datetime import timedelta +import hashlib +import json +import re +from urllib.parse import quote +import yaml + +from activity_feed import CollectionError, instant, short, stamp + +STATES = {'active', 'ready', 'running', 'blocked', 'parked', 'needs-decision', 'needs-slicing', 'complete', 'superseded'} + + +def file_at(api, repo, path, revision): + if not re.fullmatch(r'[A-Za-z0-9_./-]+', path) or '..' in path.split('/'): + raise CollectionError('invalid Pulse source path') + data = api.get(f'/repos/{repo}/contents/{path}?ref={quote(revision, safe="")}') + if data.get('type') != 'file' or data.get('size', 0) > 8000000: + raise CollectionError('unsupported Pulse source file') + if data.get('encoding') == 'none' and re.fullmatch(r'[0-9a-f]{40}', data.get('sha', '')): + # GitHub's contents API omits inline content above 1MB. Fetch the exact + # blob named at the pinned revision, never a moving download URL. + data = api.get(f'/repos/{repo}/git/blobs/{data["sha"]}') + if data.get('encoding') != 'base64' or data.get('size', 0) > 8000000: + raise CollectionError('unsupported Pulse blob') + return base64.b64decode(data['content']).decode('utf-8') + + +def fingerprint(value): + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()[:24] + + +def collect_pulse(api, repo, ref, public, identities, previous, now): + """Return a normalized campaign cache. Missing time/association is a gap.""" + if repo not in public: + return {'status': 'unavailable', 'reason': 'Pulse is not a verified public source', 'rows': []} + try: + revision = api.get(f'/repos/{repo}/commits/{quote(ref, safe="")}')['sha'] + if not re.fullmatch(r'[0-9a-f]{40}', revision): + raise ValueError('invalid revision') + ledger = yaml.safe_load(file_at(api, repo, 'campaigns.yaml', revision)) + registry = yaml.safe_load(file_at(api, repo, 'registry.yaml', revision)) + if ledger.get('version') != 1 or registry.get('schema') != 1: + raise ValueError('unsupported source version') + reviewed = instant(ledger['last_checkin']) + if reviewed > now + timedelta(minutes=5): + raise ValueError('future check-in') + campaigns = ledger['campaigns'] + tasks = ledger['tasks'] + if not isinstance(campaigns, list) or not isinstance(tasks, list) or len(campaigns) > 100: + raise ValueError('invalid campaign list') + old = {r['id']: r for r in previous.get('rows', [])} + rows, ids = [], set() + for campaign in campaigns: + cid = campaign['id'] + if not isinstance(cid, str) or not re.fullmatch(r'[a-z0-9_-]+', cid) or cid in ids or campaign['status'] not in STATES: + raise ValueError('invalid campaign identity/status') + ids.add(cid) + # Reviewed dates, render times and overall check-in stamps are not a + # meaningful change to a campaign. Track facts and recorded direction. + facts = {k: campaign.get(k) for k in ('title', 'status', 'next', 'evidence')} + facts['tasks'] = sorted(({k: t.get(k) for k in ('id', 'title', 'status', 'next', 'evidence')} + for t in tasks if t.get('campaign') == cid), key=lambda t: str(t['id'])) + task_marks = {str(t['id']): fingerprint(t) for t in facts['tasks']} + digest = fingerprint(facts) + key = f'pulse:campaign:{cid}' + prior = old.get(key, {}) + events = list(prior.get('events', [])) + # Do not fabricate a newer event from an out-of-date snapshot. + if prior.get('reviewed') and reviewed < instant(prior['reviewed']): + raise ValueError('older campaign evidence') + if digest != prior.get('fingerprint'): + changed_tasks = [t for t in facts['tasks'] if prior.get('task_marks', {}).get(str(t['id'])) != task_marks[str(t['id'])]] + summary = 'Check-in: ' + campaign['status'].replace('-', ' ') + ' · ' + str(campaign.get('next', '')) + if prior and changed_tasks: + task = changed_tasks[0] + summary = 'Task update: ' + str(task.get('title') or task['id']) + ' · ' + str(task.get('status')) + if len(changed_tasks) > 1: + summary += f' (+{len(changed_tasks) - 1} more)' + # Include check-in time in ID: a genuine later return to an old + # state is a new transition; an unchanged check-in is not. + events.append({'id': key + ':' + fingerprint([digest, stamp(reviewed)]), + 'time': stamp(reviewed), 'kind': 'checkin', + 'summary': short(summary), + 'url': f'https://github.com/{repo}/blob/{revision}/campaigns.yaml'}) + rows.append({'id': key, 'repo': repo, 'board': 'pulse', + 'board_url': f'https://{repo.split("/")[0].lower()}.github.io/PyAutoPulse/', + 'title': short(campaign['title']), 'state': campaign['status'], + 'fingerprint': digest, 'task_marks': task_marks, 'reviewed': stamp(reviewed), 'events': events}) + by_campaign = {r['id'].rsplit(':', 1)[1]: r for r in rows} + # Pulse may explicitly associate a project run with a campaign without + # requiring changes to a producer's versioned summary schema. + associations = {} + for campaign in campaigns: + for run in campaign.get('runs', []): + project = identities.get(run['repo']) + key = (project, run['run_id']) + if key in associations and associations[key] != campaign['id']: + raise ValueError('ambiguous run association') + associations[key] = campaign['id'] + undated = unassigned = dated = 0 + gaps = [] + for instance in registry.get('instances', []): + project = identities.get(instance['repo']) + if not project or project not in public: + gaps.append('project source unavailable') + continue + snapshot = json.loads(file_at(api, repo, f'snapshots/{instance["instance"]}.json', revision)) + if snapshot.get('github') != project or not re.fullmatch(r'[0-9a-f]{40}', snapshot.get('commit', '')): + raise ValueError('snapshot identity mismatch') + summary = snapshot['summary'] + runs = {} + for record in summary.get('records', []): + measured = record.get('provenance', {}).get('measured_at') + if not measured: + undated += 1 + continue + measured_time = instant(measured) + if measured_time > now + timedelta(minutes=5): + raise ValueError('future measurement') + cid = associations.get((project, record.get('run_id')), record.get('campaign_id')) + if cid not in by_campaign or not record.get('run_id'): + unassigned += 1 + continue + # Association must be producer-declared; never infer from timing, + # filenames, setup resemblance or a project-wide render timestamp. + key = (cid, record['run_id']) + run = runs.setdefault(key, {'time': measured_time, 'count': 0}) + run['time'] = max(run['time'], measured_time) + run['count'] += 1 + dated += 1 + for (cid, run_id), run in runs.items(): + row = by_campaign[cid] + eid = 'pulse:run:' + fingerprint([project, cid, run_id]) + e = {'id': eid, 'time': stamp(run['time']), 'kind': 'measurement', + 'summary': f'Measurements recorded · {run["count"]} records · acceptance not inferred', + 'url': f'https://github.com/{repo}/blob/{revision}/snapshots/{instance["instance"]}.json'} + row['events'] = [x for x in row['events'] if x['id'] != eid] + [e] + cutoff = now - timedelta(days=30) + for row in rows: + row['events'] = sorted((e for e in row['events'] if instant(e['time']) >= cutoff), key=lambda e: (e['time'], e['id']), reverse=True)[:100] + if undated: + gaps.append(f'{undated} measurement records lack event times') + if unassigned: + gaps.append(f'{unassigned} dated records lack explicit campaign/run association') + return {'status': 'partial' if gaps else 'ok', 'reason': '; '.join(gaps) or None, + 'checked': stamp(now), 'revision': revision, 'rows': rows, + 'dated_records': dated, 'undated_records': undated, 'unassigned_records': unassigned} + except (CollectionError, ValueError, KeyError, TypeError, yaml.YAMLError): + return dict(previous, status='stale', reason='Pulse evidence unavailable or invalid') + + +def publish(cache, now): + rows = [] + for row in cache.get('rows', []): + events = [e for e in row['events'] if instant(e['time']) >= now - timedelta(days=30)][:20] + if not events: + continue + rows.append({k: row[k] for k in ('id', 'repo', 'board', 'board_url', 'title', 'state')} | + {'events': events, 'time': events[0]['time'], 'history_count': len(row['events']), 'source_status': cache['status']}) + return rows diff --git a/tests/test_activity_feed.py b/tests/test_activity_feed.py index 4a389dd..5920680 100644 --- a/tests/test_activity_feed.py +++ b/tests/test_activity_feed.py @@ -147,7 +147,7 @@ def test_server_snapshot_escapes_and_keeps_instance_out(): rendered = af.render_html(feed) assert '