diff --git a/docs/demos/scheduled-calendar.log b/docs/demos/scheduled-calendar.log new file mode 100644 index 0000000..07673bf --- /dev/null +++ b/docs/demos/scheduled-calendar.log @@ -0,0 +1,7 @@ +Script started on 2026-10-20 22:32:56-04:00 [COMMAND="python3 scripts/demo-scheduled-calendar.py" ] +2026-10-26T13:17:00+00:00 old-fixed-policy= True calendar-policy= True +2026-11-02T14:17:00+00:00 old-fixed-policy= False calendar-policy= True +2026-11-02T13:17:00+00:00 old-fixed-policy= True calendar-policy= False +Synthetic local policy only; not installed admission or runtime acceptance. + +Script done on 2026-10-20 22:32:56-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/scheduled-calendar.timing b/docs/demos/scheduled-calendar.timing new file mode 100644 index 0000000..bfbbd2d --- /dev/null +++ b/docs/demos/scheduled-calendar.timing @@ -0,0 +1,2 @@ +0.069563 145 +0.000013 150 diff --git a/docs/scheduled-runs.md b/docs/scheduled-runs.md index cd71a0e..8e063b4 100644 --- a/docs/scheduled-runs.md +++ b/docs/scheduled-runs.md @@ -59,6 +59,17 @@ The trusted local config has exactly `binding`, `host`, `public_key`, `max_lateness_s`. State root must be absolute and persistent, with a provisioned existing parent. The port creates at most the root itself and fsyncs its parent entry as well. Cadence is a fixed interval of at least sixty seconds with an explicit activation boundary. +Alternatively, replace only `interval_s` with `weekday` (Monday=0..Sunday=6), +`local_time` (HH:MM) and `timezone` (IANA identifier). These two exact field +sets cannot be mixed. Calendar anchors must match the pinned weekday/time and +be unambiguous. Each integer UTC slot is converted into that zone, checked +against the local weekday/time, and round-tripped with both fold values. +Gaps and both occurrences of an overlap refuse; no time shifting or first-fold +selection. The activation boundary is inclusive. For Monday09:17 New York, +October26 2026 is13:17Z and November2 is14:17Z,608400 seconds later. +The runtime owner must bind those exact calendar qualifiers in the eligibility +query as well as the pinned configuration. This adds admission, not a timer or +a calendar-aware history/deadline reader; those remain receiver/reader contracts. No past-slot bootstrap is implied. The eligibility SELECT is compiled and reviewed by the runtime owner, stored diff --git a/scripts/demo-scheduled-calendar.py b/scripts/demo-scheduled-calendar.py new file mode 100644 index 0000000..4f6bcc9 --- /dev/null +++ b/scripts/demo-scheduled-calendar.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Private policy demonstration; no keys, graph, registration or children.""" +from datetime import datetime +from pathlib import Path +import sys +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from shuttle.scheduled_cli import cadence + +anchor = int(datetime.fromisoformat('2026-10-26T13:17:00+00:00').timestamp()) +common = dict(binding={}, host='urn:host', public_key='synthetic', registration='urn:registration', + identity_graph='urn:identity', state_root='/unused', declaration_graph='urn:jobs', + eligibility_query='SELECT ?eligible WHERE {} LIMIT 2', anchor_s=anchor, + effective_from_s=anchor, max_lateness_s=60) +fixed = cadence(dict(common, interval_s=604800)) +weekly = cadence(dict(common, weekday=0, local_time='09:17', timezone='America/New_York')) +for value in ('2026-10-26T13:17:00+00:00', '2026-11-02T14:17:00+00:00', + '2026-11-02T13:17:00+00:00'): + slot = int(datetime.fromisoformat(value).timestamp()) + print(value, 'old-fixed-policy=', fixed(slot), 'calendar-policy=', weekly(slot)) +print('Synthetic local policy only; not installed admission or runtime acceptance.') diff --git a/shuttle/scheduled_cli.py b/shuttle/scheduled_cli.py index 1c9013b..b0cd0f0 100644 --- a/shuttle/scheduled_cli.py +++ b/shuttle/scheduled_cli.py @@ -5,6 +5,8 @@ No child command, scheduler, registration or automatic key creation is provided. """ from dataclasses import dataclass +from datetime import datetime, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import json import re from pathlib import Path @@ -42,6 +44,65 @@ def _json(raw): return value +def cadence(config): + """Validate pinned local policy before any graph or key access. + + Existing fixed configurations retain their exact field contract. Calendar + configurations replace interval_s with weekday/local_time/timezone. + """ + common = {'binding', 'host', 'public_key', 'registration', 'identity_graph', + 'state_root', 'declaration_graph', 'eligibility_query', + 'anchor_s', 'effective_from_s', 'max_lateness_s'} + fixed = set(config) == common | {'interval_s'} + weekly = set(config) == common | {'weekday', 'local_time', 'timezone'} + if not (fixed or weekly): + raise ScheduledRefused('complete single-form pinned configuration required') + anchor, effective = (config[k] for k in ('anchor_s', 'effective_from_s')) + if (type(anchor) is not int or not 0 <= anchor <= 253402214400 or + type(effective) is not int or not anchor <= effective <= 253402214400): + raise ScheduledRefused('bounded cadence anchor and activation boundary required') + if fixed: + interval = config['interval_s'] + if type(interval) is not int or interval < 60: + raise ScheduledRefused('explicit fixed cadence required') + return lambda slot: (type(slot) is int and effective <= slot <= 253402214400 and + (slot-anchor) % interval == 0) + weekday, local_time, zone_name = (config[k] for k in ('weekday', 'local_time', 'timezone')) + if (type(weekday) is not int or not 0 <= weekday <= 6 or + type(local_time) is not str or + not re.fullmatch(r'(?:[01][0-9]|2[0-3]):[0-5][0-9]', local_time) or + type(zone_name) is not str or not 0 < len(zone_name) <= 255): + raise ScheduledRefused('weekly weekday/HH:MM/IANA zone required') + try: + zone = ZoneInfo(zone_name) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ScheduledRefused('valid IANA zone required') from exc + hour, minute = map(int, local_time.split(':')) + + def matches(slot): + if type(slot) is not int or not 0 <= slot <= 253402214400: + return False + try: + local = datetime.fromtimestamp(slot, zone) + if (local.weekday() != weekday or local.hour != hour or + local.minute != minute or local.second != 0 or local.microsecond != 0): + return False + naive = local.replace(tzinfo=None) + candidates = set() + for fold in (0, 1): + utc = naive.replace(tzinfo=zone, fold=fold).astimezone(timezone.utc) + if utc.astimezone(zone).replace(tzinfo=None) == naive: + candidates.add(int(utc.timestamp())) + # Neither gap shifting nor choosing one occurrence of a fold. + return candidates == {slot} + except (OverflowError, OSError, ValueError): + return False + + if not matches(anchor): + raise ScheduledRefused('anchor does not match unambiguous weekly policy') + return lambda slot: type(slot) is int and slot >= effective and matches(slot) + + def run(args): # Read bounded input first; an invalid delivery cannot touch the config, # a key, a local outbox, or the graph. No credentials are printed. @@ -57,23 +118,14 @@ def run(args): slot = packet['scheduled_at'] with Path(args.config).open() as config_file: config = _json(config_file.read(65537)) - fields = {'binding', 'host', 'public_key', 'registration', 'identity_graph', - 'state_root', 'declaration_graph', 'eligibility_query', - 'interval_s', 'anchor_s', 'effective_from_s', 'max_lateness_s'} - if set(config) != fields: - raise ScheduledRefused('complete pinned local configuration required') + matches = cadence(config) graph = _iri(config['declaration_graph']) query = config['eligibility_query'] if (type(query) is not str or not query.lstrip().upper().startswith('SELECT ') or not re.search(r'\bLIMIT\s+2\s*$', query, re.IGNORECASE)): raise ScheduledRefused('reviewed declaration SELECT required') - interval, anchor, effective = (config[k] for k in ('interval_s', 'anchor_s', 'effective_from_s')) - if (type(interval) is not int or interval < 60 or type(anchor) is not int or anchor < 0 or - type(effective) is not int or effective < anchor): - raise ScheduledRefused('explicit fixed cadence and activation boundary required') - def eligible(binding, scheduled_at): - if (scheduled_at < effective or (scheduled_at-anchor) % interval != 0): + if not matches(scheduled_at): return False rows = _rows(qc.query(query, graph=graph), 2) return (len(rows) == 1 and set(rows[0]) == {'eligible'} and diff --git a/tests/test_scheduled.py b/tests/test_scheduled.py index c559369..1b1fbbf 100644 --- a/tests/test_scheduled.py +++ b/tests/test_scheduled.py @@ -155,6 +155,24 @@ def factory(*args, **kwargs): self.assertEqual({'action': action, 'result': 'verified'}, json.loads(out.getvalue())) self.assertEqual(18, len(self.facts)) + def test_calendar_cli_preserves_real_signing_and_publication(self): + from datetime import datetime, timezone + from shuttle import scheduled_cli + # Reuse the actual four-phase signing/publication driver with only the + # pinned policy changed. Temporary keys and mocked graph stay explicit. + original = scheduled_cli._json + local = datetime.fromtimestamp(self.slot, timezone.utc) + + def calendar(raw): + value = original(raw) + if 'interval_s' in value: + value.pop('interval_s') + value.update(weekday=local.weekday(), local_time=local.strftime('%H:%M'), timezone='UTC') + return value + + with mock.patch.object(scheduled_cli, '_json', side_effect=calendar): + self.test_actual_cli_protocol_with_real_signing_and_mocked_remote() + def test_cli_malformed_or_oversized_packet_cannot_touch_config_or_graph(self): import io from shuttle import cli diff --git a/tests/test_scheduled_calendar.py b/tests/test_scheduled_calendar.py new file mode 100644 index 0000000..4c8f69c --- /dev/null +++ b/tests/test_scheduled_calendar.py @@ -0,0 +1,101 @@ +"""Private policy tests: no real keys, graph or child commands.""" +from datetime import datetime +import io +import json +from pathlib import Path +import tempfile +import unittest +from unittest import mock +from contextlib import redirect_stdout + +from shuttle import scheduled_cli as cli +from shuttle.scheduled import ScheduledRefused + + +def epoch(value): + return int(datetime.fromisoformat(value).timestamp()) + + +def config(): + return dict(binding={}, host='urn:host', public_key='synthetic', + registration='urn:registration', identity_graph='urn:identity', + state_root='/unused', declaration_graph='urn:jobs', + eligibility_query='SELECT ?eligible WHERE {} LIMIT 2', + anchor_s=epoch('2026-10-26T13:17:00+00:00'), + effective_from_s=epoch('2026-10-26T13:17:00+00:00'), + max_lateness_s=60, weekday=0, local_time='09:17', timezone='America/New_York') + + +class CalendarTests(unittest.TestCase): + def test_actual_process_paired_dst_admission(self): + policy = config() + scope = dict(job='urn:job', revision='v1', host='urn:host', + manifest_sha256='a'*64, agent='urn:agent', workflow='urn:workflow') + queries = [] + + class Port: + def __init__(self, *args, eligible, **kwargs): + self.eligible = eligible + + def admit(self, scope, slot): + if not self.eligible(scope, slot): + raise ScheduledRefused('cadence refused') + + with tempfile.TemporaryDirectory() as root: + path = Path(root)/'config.json' + path.write_text(json.dumps(policy)) + for slot in (policy['anchor_s'], epoch('2026-11-02T14:17:00+00:00')): + packet = dict(scope=scope, scheduled_at=slot) + with mock.patch.object(cli, 'ScheduledRunPort', Port), \ + mock.patch.object(cli.qc, 'query', side_effect=lambda *a, **k: queries.append(1) or { + 'rows': [{'eligible': True}], 'count': 1, 'truncated': False}), \ + mock.patch('sys.stdin', io.StringIO(json.dumps(packet))), redirect_stdout(io.StringIO()): + self.assertEqual(cli.run(type('Args', (), dict(action='admit', config=path))()), 0) + self.assertEqual(len(queries), 2) + + def test_fall_and_spring_no_fixed_seconds(self): + policy = config() + policy['anchor_s'] = policy['effective_from_s'] = epoch('2026-03-02T14:17:00+00:00') + matches = cli.cadence(policy) + for value in ('2026-03-02T14:17:00+00:00', '2026-03-09T13:17:00+00:00', + '2026-10-26T13:17:00+00:00', '2026-11-02T14:17:00+00:00'): + self.assertTrue(matches(epoch(value))) + for value in ('2026-11-02T13:17:00+00:00', '2026-11-03T14:17:00+00:00', + '2026-11-02T14:17:01+00:00'): + self.assertFalse(matches(epoch(value))) + + def test_gap_and_both_fold_occurrences_refused(self): + for time, anchor, slots in ( + ('01:30', '2026-10-25T05:30:00+00:00', + ('2026-11-01T05:30:00+00:00', '2026-11-01T06:30:00+00:00')), + ('02:30', '2026-03-01T07:30:00+00:00', + ('2026-03-08T06:30:00+00:00', '2026-03-08T07:30:00+00:00'))): + policy = config() + policy.update(weekday=6, local_time=time, anchor_s=epoch(anchor), effective_from_s=epoch(anchor)) + matches = cli.cadence(policy) + self.assertTrue(matches(epoch(anchor))) + for value in slots: + self.assertFalse(matches(epoch(value))) + policy['anchor_s'] = policy['effective_from_s'] = epoch(slots[0]) + with self.assertRaises(ScheduledRefused): + cli.cadence(policy) + + def test_fixed_unchanged_and_invalid_forms(self): + policy = config() + for key in ('weekday', 'local_time', 'timezone'): + policy.pop(key) + policy['interval_s'] = 604800 + matches = cli.cadence(policy) + self.assertTrue(matches(policy['anchor_s'])) + self.assertFalse(matches(epoch('2026-11-02T14:17:00+00:00'))) + mutations = [dict(interval_s=604800), dict(weekday=True), dict(weekday=7), + dict(local_time='09:17:00'), dict(timezone='Not/AZone'), + dict(anchor_s=True), dict(effective_from_s=-1), dict(extra=1)] + for change in mutations: + with self.subTest(change=change), self.assertRaises(ScheduledRefused): + cli.cadence(dict(config(), **change)) + self.assertFalse(cli.cadence(config())(True)) + + +if __name__ == '__main__': + unittest.main()