fix: harden iMessage pickup timing and binding output

bin/fm-courier-pickup.py

@@ -35,7 +35,8 @@
 the vc- note and appends its check wake, which the watcher's conversation ring
 surfaces within about a second (docs/watcher-continuity.md). A capture that
 fails is retried after 5 and 20 seconds, then, unless the transport shows it
-landed, the captain is told dispatch FAILURE text. An empty transcript is not
+landed, the captain is told dispatch FAILURE text. A restart retries the pending
+capture immediately, retaining its attempt count. An empty transcript is not
 filed and he is told NOT_TEXT. With every transport call bounded at 10 s, the
 failure notice is normally queued within about 75 s of reading the message,
 and its failed stage follows the notice's sent receipt. If the notice waits
@@ -73,7 +74,8 @@
 Texts go one at a time in order: the next is published only after
 the courier's terminal result receipt for the previous one.
 If a consumed request has no result receipt, pickup republishes its identical
-saved request at most once per 60 s to recover the receipt using the same ID.
+saved request after 60 s to recover the receipt using the same ID; a last
+publication timestamp ahead of wall time makes recovery due immediately.
 A sent receipt records playback completed and
 moves the stage the reply implies (receipt/progress working, question question,
 error failed, final answer done). Stages are published as ID.stage.json in
@@ -83,8 +85,12 @@
 FM_COURIER_ROOT (default /) relocates srv/courier and etc/courier for offline
 tests; FM_COURIER_USER (default courier) names the record and receipt owner.
 The binding is FM_HOME/state/imessage/binding.json, exactly
-{conversation_id, credential, destination: "imessage"}. Startup refusals exit 1,
-usage 2; per-tick failures are logged and the loop continues.
+{conversation_id, credential, destination: "imessage"} as fm-inbox.sh
+conversation bind returned it. Startup refusals exit 1, usage 2; per-tick
+failures are logged and the loop continues. Reply reads and capture retries are
+paced on the monotonic clock, so a wall-clock step neither stalls nor rushes
+them. Wall time stamps persistent records and governs their expiry and receipt
+recovery.
 """
 
 import calendar
@@ -230,8 +236,9 @@ def epoch(text):
 
 
 class Pickup:
-    def __init__(self, home, root, courier_user, now=time.time):
-        self.home, self.now = home, now
+    def __init__(self, home, root, courier_user, now=time.time, clock=time.monotonic):
+        # clock keeps reply reads and capture retries independent of wall-clock steps.
+        self.home, self.now, self.clock = home, now, clock
         self.inbound = root / 'srv/courier/inbound'
         self.outbox = root / 'srv/courier/outbox'
         self.receipts = root / 'srv/courier/inbox'
@@ -259,6 +266,7 @@ def __init__(self, home, root, courier_user, now=time.time):
         if self.s.get('version') != 1:
             raise Refused('unsupported pickup state')
         self.next_reply = 0.0
+        self.retry_at = 0.0  # The pending capture's next attempt; a restart retries at once.
         self.quiet = {}
         # The turn this process last filed. Only this single-instance peer files
         # into the iMessage conversation, so it is the conversation's tail; a
@@ -402,13 +410,14 @@ def take(self, seq, kind, key, record):
                                  'created': record['published_at'], 'notice': None,
                                  'asked': poll['asked'], 'binding': poll['binding'], 'question': poll['question']}
             self.s['polls'].pop(record['request_id'])
-        self.s['pending'].update(attempts=0, not_before=0, turn=None, request=None, previous=None)
+        self.retry_at = 0.0
+        self.s['pending'].update(attempts=0, turn=None, request=None, previous=None)
         self.save()
 
     def file(self):
         """Try the pending capture once it is due; whether the queue moved past it."""
         entry = self.s['pending']
-        if entry['not_before'] > self.now():
+        if self.retry_at > self.clock():
             return False
         try:
             if entry['turn'] is None:
@@ -430,7 +439,7 @@ def file(self):
             self.tail = None
             entry['attempts'] += 1
             if entry['attempts'] <= len(RETRY_DELAYS):
-                entry['not_before'] = self.now() + RETRY_DELAYS[entry['attempts'] - 1]
+                self.retry_at = self.clock() + RETRY_DELAYS[entry['attempts'] - 1]
                 log('filing %s failed (attempt %d): %s' % (entry['message'], entry['attempts'], exc))
                 self.save()
                 return False
@@ -589,7 +598,8 @@ def send(self):
                          if entry['mark'] and entry['mark'][1] == 'question' else {})
             result = self.receipt(entry['id'])
             if result is None and (entry['published'] is None
-                                   or (self.now() - entry['published'] >= RECEIPT_RETRY
+                                   # A publish time ahead of now means the clock stepped back.
+                                   or (not 0 <= self.now() - entry['published'] < RECEIPT_RETRY
                                        and not os.path.lexists(self.outbox / (entry['id'] + '.json')))):
                 row = entry.get('wire')
                 if row is None:
@@ -647,8 +657,8 @@ def tick(self, force_replies=False):
             self.clear('pick')
         except (Refused, OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired) as exc:
             self.once('pick', 'inbound spool unavailable: %s' % exc)
-        if self.s['pending'] is None and (force_replies or self.now() >= self.next_reply):
-            self.next_reply = self.now() + REPLY_EVERY
+        if self.s['pending'] is None and (force_replies or self.clock() >= self.next_reply):
+            self.next_reply = self.clock() + REPLY_EVERY
             try:
                 self.replies()
                 self.clear('replies')

bin/fm_inbox_conversation.py

@@ -32,13 +32,15 @@
     therefore needs no reset of state/voice-conversation/policy.json.
 
 bind (owner): {conversation_id, authenticated_principal, destination?}. Returns
-    a random transport credential bound to this conversation and its authenticated
-    principal. destination defaults to "elevenlabs" and must be one the live
-    policy authorizes (a lab authorizes only the default); it is fixed at the
-    first bind, and publish refuses any reply for another destination, so a
+    {conversation_id, credential, destination?}: a random transport credential
+    bound to this conversation and its authenticated principal, plus the bound
+    destination when it is not the default. Transport calls carry only the
+    conversation_id and credential. destination defaults to "elevenlabs" and
+    must be one the live policy authorizes (a lab authorizes only the default);
+    it is fixed at the first bind, and publish refuses any reply for another destination, so a
     reply meant for a call never goes out as a text and a text never reaches a
     call. A conversation bound to a
-    destination other than "elevenlabs" reports it from accept and audit; one
+    destination other than "elevenlabs" reports it from bind, accept and audit; one
     bound to the default carries no such field, exactly as before it existed.
     Repeat bind by the session holding the lock is idempotent; a different
     principal or destination cannot adopt the conversation.
@@ -362,7 +364,7 @@ def bind(self):
                 self.c['destination'] = destination
             self.j['conversations'][self.cid] = self.c
             self.save()
-        return {'conversation_id': self.cid, 'credential': self.c['credential']}
+        return self.scoped({'conversation_id': self.cid, 'credential': self.c['credential']})
 
     def wake(self, key):
         env = dict(os.environ, FM_HOME=str(self.home))

tests/fm-courier-pickup-cases.py

@@ -143,7 +143,8 @@ def courier_takes(receipted=True):
 binding = conversation('bind', {'conversation_id': 'text', 'authenticated_principal': 'captain',
                                 'destination': 'imessage'})
 (home / 'state/imessage').mkdir(mode=0o700)
-(home / 'state/imessage/binding.json').write_text(json.dumps(dict(binding, destination='imessage')))
+assert binding['destination'] == 'imessage', binding
+(home / 'state/imessage/binding.json').write_text(json.dumps(binding))  # exactly as bind returned it
 for directory in (inbound, outbox, receipts, courier / 'etc/courier'):
     directory.mkdir(parents=True)
 (courier / 'etc/courier/policy.toml').write_text(
@@ -467,14 +468,17 @@ def courier_takes(receipted=True):
 inflight_receipt.unlink()
 once(clock_offset=2 * 86400 + 61)
 assert (outbox / (original['id'] + '.json')).read_bytes() == original_bytes
+courier_takes(receipted=False)
+once(clock_offset=2 * 86400 + 61 - 600)  # The wall clock stepped back 10 minutes.
+assert (outbox / (original['id'] + '.json')).read_bytes() == original_bytes
 assert [row['id'] for row in courier_takes()] == [original['id']]
 once()
 assert [row['text'] for row in courier_takes()] == ['Recovery complete.']
 once()
 assert ('receipt-recovery', 'done') in stages(), stages()
 assert all(row['delivery']['state'] == 'completed'
            for row in conversation('poll', binding)['replies'] if row['response_id'].startswith('recovery-'))
-print('PASS: a pruned receipt is recovered by byte-identical same-ID republication, limited to once per 60 s')
+print('PASS: a pruned receipt is recovered by byte-identical same-ID republication, limited to once per 60 s, and not stalled by a backward clock step')
 
 message('best-effort-poll', 'Ask with an optional native poll.')
 once()
@@ -700,12 +704,8 @@ def courier_takes(receipted=True):
 # A message nobody can file - no live session - is told the failure sentence after its retries.
 (home / 'state/.lock').write_text('99999999\n')
 message('m5', 'anyone there?')
-for _ in range(3):
+for _ in range(3):  # Each run is a new process, so its retry is due at once.
     once()
-    state = json.loads(state_path.read_text())
-    if state['pending'] is not None:
-        state['pending']['not_before'] = 0  # skip the retry wait, not the retry
-        state_path.write_text(json.dumps(state))
 assert json.loads(state_path.read_text())['pending'] is None
 assert ('m5', 'failed') not in stages(), stages()
 assert [row['text'] for row in courier_takes()] == ["I couldn't reach Firstmate."]
@@ -723,8 +723,6 @@ def courier_takes(receipted=True):
     if attempt < 2:
         assert state['pending']['words'] == 'x' * 16001
         assert state['pending']['attempts'] == attempt + 1
-        state['pending']['not_before'] = 0
-        state_path.write_text(json.dumps(state))
 assert state['pending'] is None
 assert not any(r['turn_id'] == 'imsg-m-long' for r in requests()), requests()
 assert ('m-long', 'failed') not in stages(), stages()
@@ -744,10 +742,6 @@ def courier_takes(receipted=True):
                                  'digest': 'd', 'poll_message_id': 'p3'})
 for _ in range(3):
     once()
-    state = json.loads(state_path.read_text())
-    if state['pending'] is not None:
-        state['pending']['not_before'] = 0
-        state_path.write_text(json.dumps(state))
 assert json.loads(state_path.read_text())['pending'] is None
 assert [row['text'] for row in courier_takes()] == ["I couldn't reach Firstmate."]
 once()
@@ -779,6 +773,33 @@ def courier_takes(receipted=True):
 assert second.returncode == 0
 print('PASS: each record reaches the wake queue in under a second while the pickup runs')
 
+# A backward wall-clock step while running neither stalls reading replies nor sending them.
+stepped = ('import runpy,sys,time; wall=time.time; start=time.monotonic(); '
+           'time.time=lambda: wall()-(600 if time.monotonic()-start > 1 else 0); sys.argv=sys.argv[1:]; '
+           'runpy.run_path(sys.argv[0], run_name="__main__")')
+while conversation('accept', {'conversation_id': 'text'})['dispatch']:
+    pass
+running = subprocess.Popen([sys.executable, '-c', stepped, str(pickup), 'run'], env=env, stdout=subprocess.DEVNULL,
+                           stderr=subprocess.PIPE, text=True)
+try:
+    time.sleep(2.5)  # past the 10-minute step back
+    conversation('publish', {'conversation_id': 'text', 'request_id': 'imsg-req-lat0', 'response_id': 'stepped',
+                             'sequence': 1, 'kind': 'answer', 'final': True, 'destination': 'imessage',
+                             'speech_text': 'Sent despite the clock.'})
+    started = time.monotonic()
+    while not outgoing('.json'):
+        assert time.monotonic() - started < 10, 'a backward clock step stalled the reply'
+        assert running.poll() is None, 'pickup exited'
+        time.sleep(0.1)
+    assert [row['text'] for row in courier_takes()] == ['Sent despite the clock.']
+    while ('lat0', 'done') not in stages():
+        assert time.monotonic() - started < 10, 'the sent reply never moved its stage'
+        time.sleep(0.1)
+finally:
+    running.terminate()
+    running.wait(timeout=10)
+print('PASS: after a backward wall-clock step the running pickup still reads and sends replies on its cadence')
+
 # Slow failure: a hung capture queues its notice, whose sent receipt precedes the
 # failed stage, inside the courier's 120 s PICKUP_DEADLINE from the record's rename.
 hang = open(home / 'state/voice-conversation/lock', 'a')

tests/fm-inbox-conversation-cases.py

@@ -445,6 +445,9 @@ def saved(cid='calls'):
 text_bind = {'conversation_id': 'text', 'authenticated_principal': 'captain', 'destination': 'imessage'}
 text = run('bind', text_bind)
 assert run('bind', text_bind) == text
+assert set(text) == {'conversation_id', 'credential', 'destination'} and text['destination'] == 'imessage'
+assert set(call) == {'conversation_id', 'credential'}  # a default binding is unchanged
+text = {k: text[k] for k in ('conversation_id', 'credential')}  # the transport identity
 for rebind in (dict(text_bind, destination='elevenlabs'), {'conversation_id': 'text', 'authenticated_principal': 'captain'},
                {'conversation_id': 'call', 'authenticated_principal': 'captain', 'destination': 'imessage'},
                {'conversation_id': 'fax', 'authenticated_principal': 'captain', 'destination': 'fax'}):