Fix what the relay loop does each tick: binding, durability, fairness, diagnostics, supersession

packages/codex-session-relay/docs/invariants.md

@@ -100,6 +100,11 @@ status. Every row below is implemented and carries a test; the suite is the proo
 | I-62 | No notification flood | a per-recipient minimum interval and hourly cap | implemented |
 | I-63 | Unchanged states stay quiet | a tick that changes nothing writes no journal rows | implemented |
 | I-64 | An unbounded daemon loop is not constructible | `run` requires a tick count, a deadline or a stop signal | implemented |
+| I-65 | A supervisor is bounded by the owner's intent, not by a timer | it re-reads `service.json` and the stop request between segments; `RelayDaemon.run` is unchanged, so every worker it launches is still bounded by I-64 | implemented |
+| I-66 | The current generation cannot be starved by history | candidates are filtered before the per-tick budget, the current anchor is reserved, and the remainder rotates through a persisted cursor | implemented |
+| I-67 | One parent's backlog cannot consume another parent's opportunity | selection asks which parents are eligible before asking how many rows each has, then deals a bounded share one at a time | implemented |
+| I-68 | A delivery whose generation has moved on cannot be claimed | the claim statement refuses it, so the decision cannot be overtaken between checking and acting | implemented |
+| I-69 | An outstanding send is never rewritten as terminal | suppression annotates it instead, because reconciliation refuses to promote a terminal superseded aggregate and a lost response would become unresolvable | implemented |
 
 ## Work reports and the CXC report contract
 
@@ -169,6 +174,11 @@ status. Every row below is implemented and carries a test; the suite is the proo
 | Inode ownership is not proven | a hardlink or bind mount can expose the same bytes under another authorized path, which the contract permits because it authorizes paths |
 | The JSON date-time format is unvalidated | the available validator has no working format checker, so timestamp format is unverified rather than implied |
 | Terminal turns are polled, not subscribed | the transport cannot subscribe, so automatic invocation is a bounded poll that then dispatches |
+| Scheduler fairness is not transport concurrency | the adapter serialises on one worker, so a stalled call still blocks the one behind it; what is guaranteed is that a struggling parent stops being handed the rest of the budget |
+| A live process is not a working one | health is computed from staged age, anchor poll freshness and backlog; liveness is reported separately and never counted |
 | Archive state can be unknown | an inconclusive listing withholds rather than guessing, and a later observation releases it |
 | A head commit is not observable from here | the relay cannot watch a forge, so `assert_current` enforces generation on the delivery path and takes `head_sha` only from a caller that already knows the current head. A push that changes the declared manifest is structurally a new event, because the revision hash and therefore the event id change with it; a push that changes nothing declared is not, and `_check_resubmission` is what stops an old report standing for it silently |
 | `work_reports` ships with its composite key | the schema is applied with `CREATE TABLE IF NOT EXISTS`, which never reshapes an existing table, so a store created from an intermediate revision of this change that used an event-only key cannot hold a second submission. No released version has this table, so there is nothing to migrate; a store built from such a revision is recreated rather than upgraded. The write itself no longer names a conflict target, so it does not depend on which revision created the table |
+| A store is matched to its socket by provenance, not arithmetic | a hash cannot be inverted, so a store created under a spelling we cannot guess is findable only because it recorded which socket it serves. Stores record that from now on and selection asks them before creating a canonical database. A store created before that existed says nothing and is reported under `siblingStores` rather than adopted on a guess, because adopting the wrong store is worse than reporting an ambiguity |
+| Ownership decisions are taken under the lock, not beside it | probing the daemon lock and then acting on the result are two operations, and a supervisor can start between them. `stop` and `disable` hold the lock across the decision: holding it is the proof that nothing is running and nothing can start, and failing to take it is the answer that someone is there. What this does NOT give is mutual exclusion with a supervisor that is already running - that is ownership, decided from the record |
+| These are forward fixes | per-assignment settlement does not restore claims a previous global settlement already suppressed, and capped-state annotation does not reach deliveries whose generation advanced before it existed. Historical repair is separate work with its own evidence |

packages/codex-session-relay/docs/operations.md

@@ -124,22 +124,127 @@ transport call.
 | Phase | Meaning |
 |---|---|
 | `awaiting_receipt` | the child has not produced a completion receipt yet |
+| `awaiting_send` | the receipt is collected and accepted; this relay has not reached the recipient yet |
+| `in_flight` | a send was claimed and its outcome is not yet settled |
 | `parent_busy` | the parent is mid-turn; it is never interrupted |
 | `settings_rejected` | the host would not confirm the authorized execution settings |
+| `withheld:<operation>` | refused before any transport call, naming the operation that refused |
 | `turn_accepted` | the transport started a turn |
 | `awaiting_ack` | delivered, acknowledgement outstanding |
+| `awaiting_child_receipt` | a revision request was delivered; contract v1 defines no acknowledgement for that direction, so the child answers with its next completion receipt |
 | `channel_closed` | the push channel itself is unavailable; stored, not woken |
 | `superseded` | a newer generation or revision replaced this one |
+| `superseded:<reason>` | an outstanding send a newer generation or revision replaced; its state is left alone so a lost response stays reconcilable |
 
 Each carries the most recent failed operation, its concrete error, the exact settings
 difference where there is one, and the next retry time.
 
+`status` also reports `pendingIntents`: events whose delivery was wanted and refused before a
+delivery row could exist, which a paused or unauthorized assignment produces. They have no
+phase in the table above because they have no delivery; they carry the refusal, the attempt
+count and the next retry instead. Without them the most stuck state in the system was the one
+status could not show.
+
 Health is separate from liveness. A running process with a growing observation backlog is
 reported as stalled: staged event age, when each current anchor was last successfully
 polled, and the backlog per assignment are all exposed, and a live pid is never counted as
-working.
-
-Status: planned in PR-B.
+working. An anchor whose turn is terminal with nothing staged behind it is reported as
+settled and excluded from freshness: the scheduler deliberately stops reading it, so its
+last poll cannot advance, and ageing it out would report every quiet assignment as stalled.
+New staged work on that turn makes it eligible again.
+
+Settlement is recorded per assignment. Two assignments can legitimately watch the same child
+turn, and the `observations` table is keyed by the turn alone, so it can only ever name whichever
+assignment settled it first. `assignment_settlements` carries the per-assignment fact, which is
+what the observation scheduler and this health block ask. Without it every other assignment
+on a shared turn looked permanently unsettled, was re-polled on every round and spent
+observation budget forever.
+
+So is the work itself. Staged claims are selected and settled per assignment, because a child
+thread can serve several and a claim on one of its turns belongs to exactly one of them.
+Selecting by thread alone put a paused assignment's claim into an active assignment's ring,
+and settling by turn alone let whichever assignment polled first suppress the owner's claim
+while producing no receipt of its own - so the owner's parent waited on an outcome that had
+already been discarded. An inactive assignment's staged claim is left untouched until it is
+resumed, and is excluded from backlog for the same reason: the scheduler will not process it.
+
+Each parent's delivery window rotates. The parent order decides who goes first; a persistent
+per-parent cursor decides where that parent's own window starts, and it advances only by what
+was actually attempted. Without it the window was always a parent's oldest rows, so a delivery
+that fails before changing its own state stays eligible, stays oldest and blocks every later
+delivery for that parent indefinitely.
+
+A delivery that has reached its busy or pre-send attempt cap is annotated when its generation
+advances. Once a cap sets a hold, `attempt` returns before the pre-send supersession check, so
+that is the only occasion on which such a row can ever be told its generation has moved on.
+
+Status: implemented. `status` reports the phase, the most recent failed operation with its
+error code and, for a settings rejection, the exact fields the host disagreed on, plus the
+next retry time. The field-level difference is read from the raw receipt, because the
+transport classification keeps only a code.
+
+Observation health is reported beside it: staged event ages, when each current anchor was
+last successfully polled, and the backlog per assignment. A failed read updates the attempt
+time and never the success time, so an anchor whose first read failed reads as never polled
+rather than fresh. Process liveness is reported separately and is never counted as health.
+
+## What one tick guarantees
+
+The loop is bounded, so the interesting question is not what it does but what it cannot
+starve or lose.
+
+**No anchor is left behind.** A revision can reach `dispatched` by several routes, and binding
+used to happen on only one of them, which left the generation unbound and made every later
+receipt for it refused. Binding is now a recovery over state that runs first in each tick, so
+whichever route dispatched it, the next tick repairs it and a receipt arriving in that same
+tick is accepted.
+
+**A refused queue is remembered, not lost.** Finalizing a claim, recording the observation
+that finalized it and queuing what it produced are one commit. A refusal that may not last -
+a paused relationship, a recipient not yet authorized - records a delivery intent, and
+recovery retries that intent with an exponential backoff so one permanently unqueueable event
+cannot hold a slot. Anything else rolls the whole thing back, and the next tick re-observes.
+
+Absence of a delivery row is deliberately NOT treated as evidence that delivery was wanted: a
+receipt emitted with `--no-enqueue` and an event stranded by an old generation look exactly
+the same from outside, and neither should be sent.
+
+**The current generation is always reachable.** Observation reads are capped per tick. Within
+that cap the tick serves a rotating subset of relationships rather than promising every one
+of them a read, because that promise stops being possible once the relationship count passes
+the budget. Each served relationship gets its current anchor first and then a rotating slice
+of the rest, from a cursor persisted in the database so a restart resumes the rotation.
+
+| | anchor revisit | full backlog coverage |
+|---|---|---|
+| share of two or more | every service round | `ceil(R / served) * ceil(N / (share - 1))` ticks |
+| share of one | every two service rounds | `ceil(R / served) * 2N` ticks |
+
+A candidate with nothing left to learn is dropped before the budget rather than after it,
+which is what the old prefix got wrong: past eight generations the slice was permanently the
+first eight, every one already observed, and the generation actually running was never
+selected again.
+
+An observation is also no longer treated as the end of a turn. A receipt written just after
+the completion was seen still has to be resolved, so a turn is skipped only when it has been
+observed and has no unresolved staged claim.
+
+Status: implemented.
+
+**Every parent gets a turn.** Selection asks which parents have anything to send before it
+asks how much each of them has, then takes a bounded share from each, dealt one at a time.
+A single oldest-first window let one parent's backlog take every slot. Reconciliation is
+selected the same way. A parent whose send errors or defers is skipped for the rest of that
+tick only; it reserves no capacity and creates no hold.
+
+This is scheduler fairness, not transport concurrency. The adapter serialises on one worker,
+so a stalled call still blocks the one behind it.
+
+**A stale event is stopped before the send.** A generation that has moved on invalidates
+every outcome of the previous one, whether or not the new generation has produced a revision
+yet, and the claim statement itself refuses one. An outstanding send is annotated rather than
+rewritten, so reconciliation can still settle it, and an already delivered copy keeps its
+history without being read as verification of the current head.
 
 ## What a restart preserves
 

packages/codex-session-relay/src/codex_session_relay/ack.py

@@ -23,7 +23,7 @@
     currency_of,
 )
 from .delivery import COMPLETION, REVISION
-from .errors import AckRefused, RefusalReason
+from .errors import AckRefused, RefusalReason, RelayError
 from .identity import (
     ack_proof as derive_ack_proof,
     revision_request_event_id,
@@ -592,6 +592,42 @@ def bind_dispatched_revision(self, revision_event_id: str):
             dispatch_turn_id=row["dispatch_turn_id"], source="dispatch_receipt",
         )
 
+    def bind_pending_anchors(self, *, limit: int = 50) -> list:
+        """Bind every generation still anchor_pending whose revision actually dispatched.
+
+        Binding used to be a hook on ONE path - the daemon's own new dispatch - so a revision
+        that reached dispatched any other way left its generation unbound, and by I-06 every
+        later receipt for that generation was refused. The routes that missed it are ordinary:
+        the deliver command, either reconcile promotion, and a dispatch committed in the last
+        tick before a shutdown.
+
+        Recovery over state covers all of them at once, and it repairs a generation that was
+        left pending before this existed rather than only preventing new ones. bind_anchor is
+        idempotent for the same turn and refuses a conflicting rebind (I-05), so this can
+        never move an anchor that is already bound.
+        """
+        rows = self.store.all(
+            "SELECT d.event_id FROM deliveries d"
+            "  JOIN events e ON e.event_id = d.event_id"
+            "  JOIN generations g ON g.relationship_id = e.relationship_id"
+            "   AND g.execution_generation = e.execution_generation"
+            " WHERE d.kind = ? AND d.state IN (?,?) AND d.dispatch_turn_id IS NOT NULL"
+            "   AND g.anchor_state = ?"
+            " ORDER BY d.updated_at LIMIT ?",
+            (REVISION, DISPATCHED, ACKNOWLEDGED, "anchor_pending", limit),
+        )
+        bound = []
+        for row in rows:
+            try:
+                result = self.bind_dispatched_revision(row["event_id"])
+            except RelayError:
+                # A conflicting rebind stays refused and stays reportable; it is not this
+                # pass's business to resolve, and swallowing the others would hide them.
+                continue
+            if result is not None:
+                bound.append(row["event_id"])
+        return bound
+
 
 # The host reports a turn's start as WHOLE SECONDS, while we record the send with microsecond
 # precision. A reported start of N therefore means the turn really began somewhere in

packages/codex-session-relay/src/codex_session_relay/cli.py

@@ -380,8 +380,14 @@ def cmd_emit(services, args) -> dict:
     result = {"receipt": contract_record(stored), "stage": stored.get("_stage"),
               "duplicate": stored.get("_duplicate"), "terminalProof": proof,
               "observedTurnStatus": observed_status}
-    if stored.get("_stage") == "final" and not args.no_enqueue:
-        result["delivery"] = dict(services.delivery.enqueue(event))
+    if stored.get("_stage") == "final":
+        # Whatever this event replaces stops being current the moment this one is final, and
+        # that is true whether or not anyone asked to deliver THIS one. --no-enqueue skips
+        # the queue, and the annotation used to ride on it, so a predecessor already in
+        # flight kept being reported as the current delivery.
+        services.delivery.annotate_predecessors(event)
+        if not args.no_enqueue:
+            result["delivery"] = dict(services.delivery.enqueue(event))
     return result
 
 
@@ -422,21 +428,30 @@ def cmd_deliver(services, args) -> dict:
     _require_adapter(services)
     if args.event:
         record = services.delivery.attempt(args.event, services.adapter)
+        # Every route to dispatched binds its anchor, not only the daemon's own.
+        services.ack.bind_pending_anchors()
         return {"attempt": record}
     out = []
     for row in services.delivery.eligible(now=services.clock.now(), limit=args.limit):
         out.append(services.delivery.attempt(row["event_id"], services.adapter))
+    # The bulk path dispatches revisions too, so it binds for exactly the same reason the
+    # single-event path does.
+    services.ack.bind_pending_anchors()
     return {"attempts": out}
 
 
 def cmd_reconcile(services, args) -> dict:
     _require_adapter(services)
-    return services.reconciler.reconcile_attempt(args.request_id, services.adapter)
+    outcome = services.reconciler.reconcile_attempt(args.request_id, services.adapter)
+    services.ack.bind_pending_anchors()
+    return outcome
 
 
 def cmd_recover(services, args) -> dict:
     _require_adapter(services)
-    return services.reconciler.recover_on_start(services.adapter)
+    outcome = services.reconciler.recover_on_start(services.adapter)
+    outcome["anchorsBound"] = services.ack.bind_pending_anchors()
+    return outcome
 
 
 def cmd_claim(services, args) -> dict:
@@ -660,7 +675,13 @@ def cmd_show(services, args) -> dict:
 
 
 def cmd_status(services, args) -> dict:
-    return services.delivery.snapshot(relationship_id=args.relationship)
+    payload = services.delivery.snapshot(relationship_id=args.relationship)
+    # Scoped with the deliveries. A global health block beside a filtered list invites
+    # reading another assignment's backlog as this one's.
+    payload["observation"] = services.delivery.observation_health(
+        relationship_id=args.relationship,
+    )
+    return payload
 
 
 def _scheduler_wait(clock, deadline, sleeper=None):