feat(scheduled): admit pinned weekly wall-clock slots

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" <not executed on terminal>]
+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"]

docs/demos/scheduled-calendar.timing

@@ -0,0 +1,2 @@
+0.069563 145
+0.000013 150

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

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.')

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

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

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()