From a32c0f15414e5f44b242903d75dfa4ab5e2a1b74 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 6 Sep 2026 19:32:26 +0200 Subject: [PATCH] Prepare bounded telemetry runtime jobs and activity definitions Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883 --- .../telemetry-platform-ingest.md | 29 +++++ .../telemetry-platform-watchdog.md | 29 +++++ docs/signal-contract.md | 29 +++++ ...6-09-06-runtime-integration-preparation.md | 19 +++ scripts/runtime.py | 119 ++++++++++++++++++ tasks/telemetry-platform-ingest.md | 16 +++ tasks/telemetry-platform-watchdog.md | 14 +++ tests/test_runtime.py | 65 ++++++++++ workplans/RTEL-WP-0002-signal-contract.md | 9 ++ 9 files changed, 329 insertions(+) create mode 100644 activity-definitions/telemetry-platform-ingest.md create mode 100644 activity-definitions/telemetry-platform-watchdog.md create mode 100644 history/2026-09-06-runtime-integration-preparation.md create mode 100644 scripts/runtime.py create mode 100644 tasks/telemetry-platform-ingest.md create mode 100644 tasks/telemetry-platform-watchdog.md create mode 100644 tests/test_runtime.py diff --git a/activity-definitions/telemetry-platform-ingest.md b/activity-definitions/telemetry-platform-ingest.md new file mode 100644 index 0000000..d4769dc --- /dev/null +++ b/activity-definitions/telemetry-platform-ingest.md @@ -0,0 +1,29 @@ +--- +id: telemetry-platform-ingest +name: Receive platform assurance +enabled: false +owner: railiance-telemetry +governance: custodian +status: proposed +trigger: + type: cron + cron_expression: "*/5 * * * *" + timezone: UTC + misfire_policy: skip +context_sources: [] +--- + +# Receive platform assurance + +Disabled candidate. RTEL-WP-0002-T04 requires accepted executor, runtime/storage, +recipient and independent watchdog before registration/enablement. + +```rule +id: run-bounded-telemetry-job +condition: "" +action: + task_template: tasks/telemetry-platform-ingest.md + target_repo: railiance-telemetry + priority: high + labels: [telemetry, bounded-operation] +``` diff --git a/activity-definitions/telemetry-platform-watchdog.md b/activity-definitions/telemetry-platform-watchdog.md new file mode 100644 index 0000000..77156ea --- /dev/null +++ b/activity-definitions/telemetry-platform-watchdog.md @@ -0,0 +1,29 @@ +--- +id: telemetry-platform-watchdog +name: Check platform emission absence +enabled: false +owner: railiance-telemetry +governance: custodian +status: proposed +trigger: + type: cron + cron_expression: "*/2 * * * *" + timezone: UTC + misfire_policy: skip +context_sources: [] +--- + +# Check platform emission absence + +Disabled candidate. RTEL-WP-0002-T04 requires accepted executor, runtime/storage, +recipient and independent watchdog before registration/enablement. + +```rule +id: run-bounded-telemetry-job +condition: "" +action: + task_template: tasks/telemetry-platform-watchdog.md + target_repo: railiance-telemetry + priority: high + labels: [telemetry, bounded-operation] +``` diff --git a/docs/signal-contract.md b/docs/signal-contract.md index 0b9be8d..ead7a19 100644 --- a/docs/signal-contract.md +++ b/docs/signal-contract.md @@ -70,3 +70,32 @@ and retention, independent receiver-watchdog delivery, and controlled failure / producer-absence delivery acknowledged by that operator. Request scope already covers reference implementation; no live scheduler or messaging authorization is inferred. RPF-WP-0036-T04 stays open until those receipts exist. + +## Runtime job candidates + +`scripts/runtime.py` supplies bounded `ingest-report`, `check`, `health` and +`snapshot` jobs. It requires an owned 0700 state directory, uses a private umask, +and refuses a symlink database. Identical reports derive identical event IDs, +so process retries neither duplicate notices nor renew producer time. A stale +report fails instead of being stamped with the current time. + +`check` atomically records invocation time and pending notice count, including +missing-emission results. `health` reads that receipt without touching SQLite or +renewing it; older than 180 seconds is stale (exit 1), absent/invalid is an error +(exit 2). This 180-second budget is a proposed runtime default. Run this probe +from an independent executor/failure domain; local implementation alone does +not establish that independence or route an alert. + +`snapshot ` uses SQLite's backup API and validates the resulting DB. +It preserves contract binding and pending notices, refuses overwrite, and emits +SHA-256 plus counts. The receipt explicitly says off_host=false: off-host backup, +expiry, storage capacity and restored-runtime acceptance remain T04 gates. + +Candidate activity definitions are disabled: ingest every five minutes and +producer watchdog every two minutes, UTC/skip misfires. Activity-core's native +parser accepts both. The tasks describe deterministic execution, not permission +to launch coding agents periodically. Before projecting/enabling, bind the actual +executor/profile, immutable bundle, upstream platform report command, state +location and notification channel. No resolver, live schedule or second host +cron has been introduced. Strict profile-enforcing environments need the accepted +profile binding before registration; none is invented in these candidates. diff --git a/history/2026-09-06-runtime-integration-preparation.md b/history/2026-09-06-runtime-integration-preparation.md new file mode 100644 index 0000000..38bfdde --- /dev/null +++ b/history/2026-09-06-runtime-integration-preparation.md @@ -0,0 +1,19 @@ +# Runtime integration preparation — 2026-09-06 + +Continued RTEL-WP-0002-T04. Activity-core's recurring-automations playbook places +cadence in domain definitions and execution in an accepted claimant/domain +executor; no workstation cron or new scheduler is needed. Implemented a bounded +private runtime command, stable report retry IDs, watchdog receipts and a separate +read-only age probe, and SQLite-native verified snapshots. Snapshot restore tests +preserve pending operator notices and the contract binding. + +Added two disabled activity definitions (five-minute ingest, two-minute emission +check), executor instructions, and explicit independent watcher requirements. +Activity-core's own parse_file accepted both. No strict-runtime profile, package +name, recipient acceptance or deployed claim executor was invented. + +Validation: 14 tests passed; native activity parser accepted both disabled +candidates. No live schedule was registered, no notification sent and no runtime +or snapshot proof was promoted to off-host recovery. Requested operator/channel +selection before implementing the actual notification destination. T04 remains +waiting for that selection and accepted execution/storage/independence bindings. diff --git a/scripts/runtime.py b/scripts/runtime.py new file mode 100644 index 0000000..c5fbdb0 --- /dev/null +++ b/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()) diff --git a/tasks/telemetry-platform-ingest.md b/tasks/telemetry-platform-ingest.md new file mode 100644 index 0000000..77c0149 --- /dev/null +++ b/tasks/telemetry-platform-ingest.md @@ -0,0 +1,16 @@ +# Receive one platform assurance report + +Prerequisites: RTEL-WP-0002-T04 accepted private executor, immutable source +revision, state directory and upstream platform capture/evaluate command. Do +not infer these from the candidate definition or launch a coding agent per tick. + +The platform owner supplies a fresh metadata-only evaluator report. In the +approved executor run `python3 scripts/runtime.py --contract +contracts/platform-assurance.json --state-dir +ingest-report ` from the pinned telemetry bundle. + +Record the bounded result on the claimed activity-core ops run and complete or +fail it. Preserve the identical report for transport retries; runtime derives +a stable event ID. Never replace its timestamp with execution time or treat +missing upstream output as healthy. No arbitrary shell arguments or recipient +addresses are supplied through event data. This job sends no notification. diff --git a/tasks/telemetry-platform-watchdog.md b/tasks/telemetry-platform-watchdog.md new file mode 100644 index 0000000..a117754 --- /dev/null +++ b/tasks/telemetry-platform-watchdog.md @@ -0,0 +1,14 @@ +# Check producer absence independently of ingestion + +Prerequisites: RTEL-WP-0002-T04 accepted private executor and persistent state. +Run `python3 scripts/runtime.py --contract contracts/platform-assurance.json +--state-dir check` from the pinned telemetry bundle. +This records missing emission in the local inbox even if no event was received. +Record the result on the claimed activity-core ops run. Do not acknowledge +notices automatically. Dispatch to the actual operator only through the accepted +notification adapter after recipient/channel authorization. + +A separate failure domain must invoke the read-only `health` command and route +stale/missing/unreachable receiver results. That watcher cannot depend solely on +this same scheduler, host or receiver database. Its deployment and delivery proof +are still T04 acceptance gates. No second workstation cadence is installed. diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..ce7d917 --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,65 @@ +from datetime import datetime, timezone, timedelta +import json +from pathlib import Path +import sys +import tempfile +import unittest +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) +from receiver import Receiver +from runtime import ingest_report, private_directory, snapshot, watchdog_health + +NOW = datetime(2026, 9, 6, tzinfo=timezone.utc) + + +class RuntimeTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) + self.directory = Path(self.temp.name) + self.contract = dict(schema='railiance-telemetry.stream.v1', stream='platform', + producer='railiance-platform', recipient='test-operator', signals=['db.restore'], + max_event_age_seconds=900, heartbeat_seconds=900, retention_days=30) + self.receiver = Receiver(self.directory / 'receiver.db', self.contract) + self.addCleanup(self.receiver.close) + self.report = dict(schema='railiance-platform.assurance-signal.v1', + cluster_uid='a553c742-0115-43d4-99a4-a5ca56fe0786', evaluated_at=NOW.isoformat(), + signals={'db.restore': {'state': 'failed', 'owner': 'railiance-platform'}}, + transport='unmonitored', guarantees='unsupported', threshold_status='local-diagnostic-only', healthy=False) + + def test_report_retry_survives_receiver_restart(self): + first = ingest_report(self.receiver, self.report, NOW) + second = Receiver(self.directory / 'receiver.db', self.contract) + try: + retry = ingest_report(second, self.report, NOW + timedelta(seconds=100)) + self.assertEqual(first['id'], retry['id']) + self.assertEqual(retry['status'], 'duplicate') + self.assertEqual(len(second.inbox()['notices']), 1) + self.assertEqual(second.check(NOW + timedelta(seconds=901))['state'], 'missing-emission') + finally: second.close() + + def test_old_report_is_not_renewed(self): + with self.assertRaises(ValueError): + ingest_report(self.receiver, self.report, NOW + timedelta(seconds=901)) + + def test_snapshot_restores_pending_notice_and_contract_binding(self): + ingest_report(self.receiver, self.report, NOW) + path = self.directory / 'snapshot.db' + proof = snapshot(self.receiver, path) + self.assertEqual(proof['pending_notices'], 1) + self.assertFalse(proof['off_host']) + restored = Receiver(path, self.contract) + try: self.assertEqual(restored.inbox(), self.receiver.inbox()) + finally: restored.close() + with self.assertRaises(FileExistsError): snapshot(self.receiver, path) + + def test_shared_or_symlink_directory_refused(self): + shared = self.directory / 'shared'; shared.mkdir(mode=0o755) + with self.assertRaises(ValueError): private_directory(shared) + link = self.directory / 'link'; link.symlink_to(self.directory, target_is_directory=True) + with self.assertRaises(ValueError): private_directory(link) + + def test_watchdog_check_does_not_renew_its_receipt(self): + receipt = dict(state='receiving', recipient='test-operator', delivery='local-inbox-only', pending_notices=0, checked_at=NOW.isoformat()) + self.assertEqual(watchdog_health(receipt, NOW)['state'], 'watchdog-current') + self.assertEqual(watchdog_health(receipt, NOW + timedelta(seconds=181))['state'], 'watchdog-stale') + self.assertEqual(receipt['checked_at'], NOW.isoformat()) + with self.assertRaises(ValueError): watchdog_health(receipt, NOW - timedelta(seconds=1)) diff --git a/workplans/RTEL-WP-0002-signal-contract.md b/workplans/RTEL-WP-0002-signal-contract.md index 93ed199..9943dd1 100644 --- a/workplans/RTEL-WP-0002-signal-contract.md +++ b/workplans/RTEL-WP-0002-signal-contract.md @@ -77,3 +77,12 @@ operator, and receiver/scheduler failure is independently detectable. Preserve receipts across restart; demonstrate capacity/retention handling. Only these receipts can satisfy RPF-WP-0036-T04. No package repo name or deployment grant is invented here. This live task holds the residual explicitly. + +T04 implementation follow-up, September 6: added deterministic runtime jobs, +restart-stable report ingestion, an independently invocable watchdog-age probe, +and verified SQLite snapshots preserving pending notices. Added disabled domain +activity definitions and executor task contracts following activity-core's +recurring-automations playbook. Native parser accepts both; 14 tests pass. +Recipient/channel clarification is pending. Runtime/profile and failure-domain +binding, off-host custody and actual notification acknowledgment remain unproven; +T04 remains wait, with no live schedule enabled.