fix(events): three defects the live rig found, two of them data loss

The whole-rig walk (ServUO + sidecar + website) against a real two-phase event.

- **A WS reconnect would have orphaned every live resource.** The backfill
  replays the last several `server.hello` frames in order — this rig saw three,
  each with a different `bootId` — so every replayed frame reads as a restart,
  and the intermediate ones compare a resource stamped with the CURRENT boot
  against a boot that ended hours ago. The row is then `orphaned`: a live crier
  line core will never take down again, lost to nothing worse than the website
  reconnecting. Gated on `!fromBackfill`, the rule the engagement fan-out and
  the SSE broadcast beside it already state. The website-was-down case is not
  missed — core asks every module at its own boot.
- **The shard explains its refusals and the run log dropped the explanation.**
  A 403 body reads `{"reason":"admin write plane disabled"}`; `legError` looks
  for `data.message`, finds nothing, and reports "sidecar responded 403". For a
  staff member clicking a button that is survivable. For an event that ran at
  four in the morning the run log is the only place anyone will learn why.
- **The "not retried" clause explained the wrong thing on a permanent status.**
  A 403 will not succeed on any attempt, so telling an operator it was not
  retried "because a repeat would announce twice" points them at a policy
  decision instead of at the switch they have to flip. The clause is now added
  only where a retry was genuinely given up, and 403/404 join the statuses the
  keyed verbs treat as terminal.

Co-Authored-By: Claude <noreply@anthropic.com>

server/test/shardIngest.eventReconcile.test.js

@@ -87,6 +87,35 @@ test('a hello with no bootId at all changes nothing', async () => {
   assert.ok(!deps.order.includes('reconcile'))
 })
 
+test('a backfill replay never reconciles, however many boots it walks through', async () => {
+  // **The defect the live rig found, and nothing else could.** A WS reconnect
+  // replays the last several `server.hello` frames in order — this rig saw three,
+  // each with a different `bootId` — so every replayed frame looks like a
+  // restart. Acting on the intermediate ones would compare a resource stamped
+  // with the CURRENT boot against a boot that ended hours ago and mark it
+  // `orphaned`: a live crier line core will never take down again, lost to
+  // nothing worse than the website reconnecting.
+  const deps = makeDeps()
+  await shardIngest.ingest(hello('boot-1'), deps)
+  for (const boot of ['boot-2', 'boot-3', 'boot-4']) {
+    await shardIngest.ingest(hello(boot), { ...deps, fromBackfill: true })
+  }
+  assert.ok(!deps.order.includes('reconcile'))
+  // The replay still moves the tracked boot on, so the NEXT live hello is
+  // measured against where the replay left off rather than against boot-1.
+  assert.ok(deps.order.includes('recordStatus:boot-4'))
+})
+
+test('a live hello after a replay is still a restart', async () => {
+  // The gate is about the frame, not about the module going quiet: skipping the
+  // replay must not make the next genuine restart invisible.
+  const deps = makeDeps()
+  await shardIngest.ingest(hello('boot-1'), deps)
+  await shardIngest.ingest(hello('boot-2'), { ...deps, fromBackfill: true })
+  await shardIngest.ingest(hello('boot-3'), deps)
+  assert.equal(deps.order.filter((s) => s === 'reconcile').length, 1)
+})
+
 test('a reconcile that throws does not take the ingest down with it', async () => {
   // Fire-and-forget by the contract, and the feed must survive one bad module:
   // `ingest()` never throws, because a single event may not kill the socket.

server/test/uoEventActions.test.js

@@ -142,15 +142,54 @@ test('a broadcast is never retried, whatever the sidecar says', async () => {
   // a shard that timed out. The last two are genuinely transient, and this is
   // the trade being taken knowingly — a lost announcement is cheaper than one
   // delivered twice to everyone online.
-  for (const status of [0, 400, 401, 409, 503, 504]) {
+  for (const status of [0, 400, 401, 403, 409, 503, 504]) {
     uoLinkClient.adminBroadcast = async () => ({ ok: false, status, error: `status ${status}` })
     const result = await broadcast.perform({ runId: 7, params: { text: 'hear ye' }, verify: false })
     assert.equal(result.ok, false)
     assert.equal(result.retry, false, `a ${status} must not be retried`)
-    assert.match(result.error, /announce twice/, 'the refusal must say why it is not retried')
+    // The clause belongs only where a retry was genuinely given up. On a
+    // permanent status it would explain the wrong thing.
+    if (!actions.PERMANENT_STATUSES.has(status)) {
+      assert.match(result.error, /announce twice/, 'a discarded retry must say why')
+    } else {
+      assert.doesNotMatch(result.error, /announce twice/, `a ${status} was never retryable`)
+    }
   }
 })
 
+test("the shard's own words reach the run log, not just a status code", async () => {
+  // **The rig found this.** The sidecar refuses a broadcast with
+  // `{"reason":"admin write plane disabled"}` and `legError` looks for
+  // `data.message`, so the run console read "sidecar responded 403" for a cause
+  // the shard had already explained in a sentence. A staff member clicking a
+  // button knows what they switched off; an event that ran at four in the morning
+  // leaves the run log as the only place anyone will learn why.
+  uoLinkClient.adminBroadcast = async () => ({
+    ok: false,
+    status: 403,
+    data: { kind: 'admin.error', reason: 'admin write plane disabled' },
+    error: 'sidecar responded 403',
+  })
+  const result = await byId('uo.broadcast').perform({ runId: 1, params: { text: 'hear ye' }, verify: false })
+  assert.match(result.error, /admin write plane disabled/)
+  // And NOT the double-announce clause: a 403 will not succeed on any attempt, so
+  // pointing an operator at a policy decision misdirects them away from the
+  // switch they actually have to flip.
+  assert.doesNotMatch(result.error, /announce twice/)
+  assert.equal(result.retry, false)
+})
+
+test('a permanent refusal of a keyed verb is not retried either', async () => {
+  // Same distinction on the other side: the keyed verbs DO retry a transient, and
+  // must not burn three attempts on a refusal that cannot change.
+  uoLinkClient.postTownCrier = async () => ({ ok: false, status: 403, data: { reason: 'admin write plane disabled' } })
+  const result = await byId('uo.towncrier.post').perform({
+    runId: 1, idempotencyKey: 'k'.repeat(40), params: { lines: 'hear ye' }, verify: false,
+  })
+  assert.equal(result.retry, false)
+  assert.match(result.error, /admin write plane disabled/)
+})
+
 test('a broadcast names its run in the shard audit, not a staff member', async () => {
   await byId('uo.broadcast').perform({ runId: 42, params: { text: 'hear ye', hue: 1153 }, verify: false })
   assert.equal(calls.broadcast.length, 1)
@@ -223,7 +262,7 @@ test('the keyed verbs DO retry, because a repeat replaces', async () => {
     const params = { lines: 'hear ye', title: 'The Fair', body: 'Merchants gather.' }
     // The announce leg's own classification of this transport, reused rather
     // than re-decided: a config or data problem is terminal, the rest transient.
-    for (const [status, retry] of [[400, false], [401, false], [409, false], [503, true], [504, true], [0, true]]) {
+    for (const [status, retry] of [[400, false], [401, false], [403, false], [409, false], [503, true], [504, true], [0, true]]) {
       uoLinkClient[stub] = async () => ({ ok: false, status, error: `status ${status}` })
       const result = await byId(id).perform({ runId: 1, idempotencyKey: 'k'.repeat(40), params, verify: false })
       assert.equal(result.ok, false)