tools: let the control guard journal recovered write losses

On q1n1's gadget-NCM link one lost datagram in a memory write surfaces
as a transport read timeout; with Q1N1_UDP_RETRIES the q1n1 Link then
resynchronises (filler, drain, NOP round trip) and repeats the write.
The guard froze the session on that first inner timeout, so the
recovery could never run under the launcher (next3g).

Inside link.writemem on a datagram transport whose Link recovers
losses, and with no native call unresolved, a transport timeout is now
journaled (transport_loss, transport_loss_recovered) instead of
freezing. A write that still fails freezes the session as before; P_
requests, reads and everything else are unchanged.

replay-a16-control-guard.py adds four datagram-loss cases (recovered,
retries off, lost on every attempt, resync NOP lost); the recovered
case fails against the previous guard.

tools/a16-control-guard.py

@@ -6,6 +6,8 @@ Use ControlGuard(proxy, record, save).proxy from the beginning of admission.
 The save callback must durably save record and must never access the target.
 The guard owns this proxy/link until close; do not retain or reuse raw handles.
 This is a single-owner session guard, not a reconnect or recovery mechanism.
+The one recovery it lets through is the q1n1 Link's own, inside a memory
+write on a datagram link (see _recoverable_write).
 
 Failure inventory and contract: docs/A16-CPU-POLICY-REVIEW-20261018.md.
 """
@@ -20,6 +22,14 @@ class Frozen(RuntimeError):
     """Target I/O is prohibited for this session."""
 
 
+def _timed_out(exc):
+    """A transport timeout as the q1n1 Link raises it (TimeoutError from a
+    datagram transport, or its ProxyTimeout), matched by name so the guard
+    keeps no import of the proxy module."""
+    return not isinstance(exc, Frozen) and any(
+        c.__name__ in ('TimeoutError', 'ProxyTimeout') for c in type(exc).__mro__)
+
+
 def _summary(value):
     if isinstance(value, (bytes, bytearray, memoryview)):
         return dict(bytes=len(value), sha256=hashlib.sha256(value).hexdigest())
@@ -56,6 +66,9 @@ class ControlGuard:
 
     Any exception from a target-facing operation freezes the session. Definite
     SCMI status errors are decoded above this layer and need not freeze it.
+    A transport timeout inside link.writemem, when the Link recovers lost
+    datagrams itself, is journaled instead; the write still freezes the
+    session if it fails as a whole.
     A returned P_CALL acknowledges execution only; the conductor must validate
     its expected return/report and call freeze() on ambiguous results.
     """
@@ -74,6 +87,8 @@ class ControlGuard:
         self._frozen = False
         self._reason = None
         self._closed = False
+        self._raw_transport = getattr(self._raw_link, 'transport', None)
+        self._write_depth = 0          # nesting of recoverable link.writemem calls
 
         # Internal helpers are bound to the raw proxy/link. Guard their lower
         # boundary too, so catching an exception inside a helper cannot revive I/O.
@@ -183,6 +198,21 @@ class ControlGuard:
             self.freeze(f'proxy.request({opcode:#x}) failed: {type(exc).__name__}: {exc}')
             raise
 
+    def _recoverable_write(self, kind, name):
+        """Whether this call is a memory write the q1n1 Link may recover itself.
+
+        On a datagram link (udplink) one lost datagram leaves the target waiting
+        for write data and the host waiting for a reply; with Q1N1_UDP_RETRIES
+        the Link then resynchronises (filler to finish the target's read, a
+        drain, and a NOP round trip that proves the stream back in step) and
+        repeats the write. A memory write is idempotent, so its inner transport
+        timeouts are not ambiguous target state. Never a P_ request, never a
+        read (a repeated MMIO read can have side effects), never while a native
+        call is unresolved."""
+        return (kind == 'link' and name == 'writemem' and self._state['native_inflight'] is None
+                and getattr(self._raw_link, 'loss_retries', 0) > 0
+                and getattr(self._raw_transport, 'datagrams', False))
+
     def _invoke(self, kind, name, operation, args, kwargs):
         if kind in ('link', 'transport') and name == 'close':
             # The link close is a host resource release, not a target command.
@@ -192,6 +222,10 @@ class ControlGuard:
         self._require_open()
         self._state['operations'] += 1
         label = f'{kind}.{name}'
+        recoverable = self._recoverable_write(kind, name)
+        if recoverable:
+            self._write_depth += 1
+            losses = self._state.get('transport_losses', 0)
         try:
             if kind == 'link' and name == 'readmem':
                 # Do not allow the link helper to retry a corrupted payload.
@@ -210,10 +244,23 @@ class ControlGuard:
                     raise RuntimeError('split raw native request bypasses journaled completion')
             result = operation(*args, **kwargs)
             self._require_open()         # also catches swallowed inner failures
+            if recoverable and self._state.get('transport_losses', 0) != losses:
+                self._event('transport_loss_recovered', operation=label,
+                            losses=self._state['transport_losses'] - losses)
+                self._persist()
             return result
         except BaseException as exc:
+            if kind == 'transport' and self._write_depth and _timed_out(exc):
+                # Inside a recoverable write: the Link resynchronises next.
+                self._state['transport_losses'] = self._state.get('transport_losses', 0) + 1
+                self._event('transport_loss', operation=label, error=f'{type(exc).__name__}: {exc}')
+                self._persist()
+                raise
             self.freeze(f'{label} failed: {type(exc).__name__}: {exc}')
             raise
+        finally:
+            if recoverable:
+                self._write_depth -= 1
 
     def restore_scratch(self, scratch):
         """Return a host-only skipped record if native/transport state is unknown."""

tools/replay-a16-control-guard.py

@@ -222,6 +222,130 @@ def run_case(q, guard_module, name, *, phase=None, fault=None, raw=False,
     return record
 
 
+class DatagramTarget:
+    """A q1n1 target behind a datagram link (udplink), for MEMWRITE and NOP:
+    it scans for a request, checks request and data checksums, and answers
+    ST_XFRERR for a write whose data does not check out. The transport carries
+    1400-byte datagrams and loses the ones a case names. Nothing is opened."""
+
+    datagrams = True
+
+    def __init__(self, q, drop):
+        self.q, self.drop = q, set(drop)
+        self.sent = 0
+        self.stream, self.out = bytearray(), bytearray()
+        self.memory = {}
+        self.trace, self.closed, self.xfrerr = [], 0, 0
+
+    def close(self):
+        self.closed += 1
+
+    def write(self, data):
+        q = self.q
+        data = bytes(data)
+        for at in range(0, len(data), 1400):
+            self.sent += 1
+            if self.sent in self.drop:
+                self.trace.append(dict(operation='dropped', datagram=self.sent))
+                continue
+            self.stream += data[at:at + 1400]
+        self.trace.append(dict(operation='write', bytes=len(data)))
+        while True:
+            at = self.stream.find(b'\xff\x55\xaa')
+            if at < 0:
+                del self.stream[:max(0, len(self.stream) - 2)]
+                return
+            del self.stream[:at]
+            if len(self.stream) < 64:
+                return
+            request = bytes(self.stream[:64])
+            kind = struct.unpack_from('<I', request)[0]
+            if q.checksum(request[:60]) != struct.unpack_from('<I', request, 60)[0]:
+                del self.stream[:3]
+                continue
+            if kind == q.REQ_MEMWRITE:
+                address, size, expected = struct.unpack_from('<QQI', request, 4)
+                if len(self.stream) < 64 + size:
+                    return
+                payload = bytes(self.stream[64:64 + size])
+                del self.stream[:64 + size]
+                status = 0
+                if q.checksum(payload) == expected:
+                    self.memory[address] = payload
+                else:
+                    status = q.ST_XFRERR
+                    self.xfrerr += 1
+                self.reply(kind, status, struct.pack('<I', expected))
+            elif kind == q.REQ_NOP:
+                del self.stream[:64]
+                self.reply(kind, 0, struct.pack('<Q', 0))
+            else:
+                raise AssertionError(f'unexpected request {kind:#x}')
+
+    def reply(self, kind, status, data):
+        prefix = struct.pack('<Ii24s', kind, status, data.ljust(24, b'\0'))
+        self.out += prefix + struct.pack('<I', self.q.checksum(prefix))
+
+    def read(self, size, timeout):
+        self.trace.append(dict(operation='read', size=size))
+        if len(self.out) < size:
+            raise TimeoutError(f'expected {size} bytes, got {len(self.out)}')
+        data = bytes(self.out[:size])
+        del self.out[:size]
+        return data
+
+    def drain(self, quiet, limit):
+        kept = bytes(self.out)
+        self.out.clear()
+        return kept
+
+
+def run_loss_case(q, guard_module, name, *, retries, drop, expect):
+    """A memory write over a lossy datagram link: the guard must let the
+    Link's own recovery through (journaled) and still freeze when it fails."""
+    record = dict(name=name, retries=retries, dropped=sorted(drop), hardware=False)
+    snapshots = []
+    target = DatagramTarget(q, drop)
+    link = q.Link('offline-replay-only', transport=target)
+    link.loss_retries = retries
+    proxy = q.Proxy(link)
+    proxy.stage_features = lambda: 0
+    guard = guard_module.ControlGuard(proxy, record, lambda: snapshots.append(json.loads(json.dumps(record))))
+    p = guard.proxy
+    data = bytes((k * 7 + 3) & 0xff for k in range(6000))
+    try:
+        p.link.writemem(0x8000, data)
+        record['outcome'] = 'written'
+    except BaseException as exc:
+        record['raised'] = f'{type(exc).__name__}: {exc}'
+        record['outcome'] = 'frozen' if guard.frozen else 'raised'
+    events = [e['event'] for e in record['control_guard']['events']]
+    record['events'] = events
+    check(record['outcome'] == expect, f'{name}: {record["outcome"]}, expected {expect}')
+    if expect == 'written':
+        check(not guard.frozen and target.memory.get(0x8000) == data, f'{name}: write not recovered intact')
+        check('transport_loss' in events and 'transport_loss_recovered' in events,
+              f'{name}: recovery was not journaled')
+        check(any('transport_loss' in [e['event'] for e in snap['control_guard']['events']]
+                  for snap in snapshots), f'{name}: loss not saved before the recovery went on')
+        p.link.writemem(0x9000, data[:100])            # the session goes on
+        check(target.memory.get(0x9000) == data[:100], f'{name}: session did not continue')
+    else:
+        check(guard.frozen, f'{name}: a failed write did not freeze the session')
+        count = len(target.trace)
+        for attempt in (lambda: p.link.writemem(0, b'x'), lambda: p.link.transport.read(1, 0.01)):
+            try:
+                attempt()
+            except guard_module.Frozen:
+                pass
+            else:
+                raise AssertionError(f'{name}: I/O after freeze was not blocked')
+        check(len(target.trace) == count, f'{name}: target I/O occurred after freeze')
+    guard.close()
+    record['wire_trace'] = target.trace
+    return record
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--output', required=True, type=Path)
@@ -258,6 +382,19 @@ def main():
             save_json(args.output, result)
             result['cases'].append(run_case(q, guard_module, name, **options))
             save_json(args.output, result)
+        # Datagram loss in a 6000-byte memory write. Datagrams: the request (1)
+        # and its data (2-6); after a loss the resync's filler (7-11) and NOP
+        # (12); then the repeated request (13) and data (14-18).
+        loss_cases = [
+            ('udp-write-loss-recovered', dict(retries=1, drop={3}, expect='written')),
+            ('udp-write-loss-retries-off', dict(retries=0, drop={3}, expect='frozen')),
+            ('udp-write-loss-every-attempt', dict(retries=1, drop={3, 15}, expect='frozen')),
+            ('udp-write-loss-resync-nop-lost', dict(retries=1, drop={3, 12}, expect='frozen'))]
+        for name, options in loss_cases:
+            result['current_case'] = name
+            save_json(args.output, result)
+            result['cases'].append(run_loss_case(q, guard_module, name, **options))
+            save_json(args.output, result)
         result.update(status='PASS', cases_passed=len(result['cases']), finished_unix=time.time())
         result.pop('current_case', None)
     except BaseException as exc: