Prepare bounded telemetry runtime jobs and activity definitions

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/runtime.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Bounded private telemetry jobs for an external scheduler/executor."""
+import argparse
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import sqlite3
+import stat
+import sys
+import tempfile
+import uuid
+from platform_event import translate
+from receiver import Receiver, read_json, instant
+
+
+def private_directory(directory):
+    directory.mkdir(mode=0o700, parents=False, exist_ok=True)
+    info = directory.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
+        raise ValueError('private owned directory required')
+
+
+def atomic_json(path, value):
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(mode='w', dir=path.parent, delete=False) as out:
+            temporary = out.name
+            json.dump(value, out)
+            out.flush(); os.fsync(out.fileno())
+        os.replace(temporary, path)
+        temporary = None
+        fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try: os.fsync(fd)
+        finally: os.close(fd)
+    finally:
+        if temporary: os.unlink(temporary)
+
+
+def ingest_report(receiver, report, now):
+    event = translate(report, receiver.contract)
+    # Identical producer reports survive process retries without new identities.
+    fingerprint = hashlib.sha256(json.dumps(report, sort_keys=True).encode()).hexdigest()
+    event['id'] = str(uuid.uuid5(uuid.NAMESPACE_URL,
+        receiver.contract['stream'] + ':' + fingerprint))
+    return receiver.ingest(event, now)
+
+
+def snapshot(receiver, output):
+    # SQLite backup includes committed transactions; never copy an open DB file.
+    fd = os.open(output, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    os.close(fd)
+    target = sqlite3.connect(output)
+    try:
+        receiver.db.backup(target)
+        if target.execute('PRAGMA quick_check').fetchone() != ('ok',):
+            raise ValueError('snapshot integrity failed')
+        events = target.execute('SELECT COUNT(*) FROM events').fetchone()[0]
+        pending = target.execute('SELECT COUNT(*) FROM notices WHERE acknowledged IS NULL').fetchone()[0]
+    finally:
+        target.close()
+    with output.open('rb') as source:
+        os.fsync(source.fileno())
+        digest = hashlib.file_digest(source, 'sha256').hexdigest()
+    return {'status': 'local-snapshot-verified', 'sha256': digest,
+            'events': events, 'pending_notices': pending, 'off_host': False}
+
+
+def watchdog_health(receipt, now):
+    if set(receipt) != {'state', 'recipient', 'delivery', 'pending_notices', 'checked_at'}:
+        raise ValueError('watchdog receipt fields')
+    age = (now - instant(receipt['checked_at'])).total_seconds()
+    if age < 0: raise ValueError('future watchdog receipt')
+    return {'state': 'watchdog-current' if age <= 180 else 'watchdog-stale',
+            'delivery': 'local-check-only'}
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--contract', required=True, type=Path)
+    p.add_argument('--state-dir', required=True, type=Path)
+    sub = p.add_subparsers(dest='command', required=True)
+    sub.add_parser('ingest-report').add_argument('report', type=Path)
+    sub.add_parser('check')
+    sub.add_parser('health')
+    sub.add_parser('snapshot').add_argument('output', type=Path)
+    a = p.parse_args()
+    receiver = None
+    os.umask(0o077)
+    try:
+        if a.command == 'health':
+            result = watchdog_health(read_json(a.state_dir / 'watchdog-receipt.json'), datetime.now(timezone.utc))
+            print(json.dumps(result))
+            return 0 if result['state'] == 'watchdog-current' else 1
+        private_directory(a.state_dir)
+        db = a.state_dir / 'receiver.db'
+        # The private directory is the trust boundary. Do not follow existing links.
+        if db.is_symlink(): raise ValueError('database symlink')
+        receiver = Receiver(db, read_json(a.contract))
+        now = datetime.now(timezone.utc)
+        if a.command == 'ingest-report': result = ingest_report(receiver, read_json(a.report), now)
+        elif a.command == 'snapshot': result = snapshot(receiver, a.output)
+        else:
+            result = receiver.check(now)
+            result['pending_notices'] = len(receiver.inbox()['notices'])
+            # A successful invocation is a heartbeat, not a declaration of health.
+            atomic_json(a.state_dir / 'watchdog-receipt.json',
+                        dict(result, checked_at=now.isoformat()))
+        print(json.dumps(result))
+        return 0
+    except (ValueError, TypeError, KeyError, AttributeError, OSError, sqlite3.Error):
+        print(json.dumps({'status': 'failed', 'error': 'runtime-unavailable-or-invalid-input'}))
+        return 2
+    finally:
+        if receiver is not None: receiver.close()
+
+
+if __name__ == '__main__': sys.exit(main())