fleetd #803/#804: remove dead mayTaskReadNudge plumbing

The #804 grant gave a collaborator TASK_READ for its own ticket, which
dissolved the only case the mayTaskReadNudge gate existed for. With the
real ownership classifier forced to t -> true at its one call site, the
grant reduced to "every non-anonymous role", so the filter could never
suppress a ticket nudge in production. Revert ReplyPushLoop, PrimaryRegistry
and FleetMcp to their pre-plumbing shape, and drop the two tests that only
proved the dead seam, not that any caller could reach it.

fleetd/src/main/java/dev/ltms/fleet/mcp/FleetMcp.java

@@ -533,13 +533,8 @@ public final class FleetMcp {
                     // scope, rather than re-derived from a bare terminal string later.
                     boolean mayDrainNudge = Authz.permits(caller, Authz.Action.DRAIN, target);
                     boolean mayAnswerNudge = Authz.permits(caller, Authz.Action.ANSWER, target);
-                    // A ticket nudge tells this same caller to poll a ticket IT created, which
-                    // TASK_READ always lets an owning caller do -- so the classifier below only
-                    // needs to report ownership as true, leaving the role check itself real.
-                    boolean mayTaskReadNudge = Authz.permits(caller, Authz.Action.TASK_READ, target,
-                            Authz.NO_KNOWN_LEAD_OR_COLLABORATOR, Authz.NO_OBSERVER_SEND_TARGET, t -> true);
                     Runnable onAccepted = () -> primaryRegistry.recordDelegation(target, callerTerminal,
-                            delegatorName, mayDrainNudge, mayAnswerNudge, mayTaskReadNudge);
+                            delegatorName, mayDrainNudge, mayAnswerNudge);
                     // wait defaults to true (block for the reply); wait:false is fire-and-poll.
                     return Boolean.FALSE.equals(a.get("wait"))
                             ? sendAsync(messages, target, content, onAccepted, workers.profiles(), caller)

fleetd/src/main/java/dev/ltms/fleet/mcp/PrimaryRegistry.java

@@ -50,13 +50,11 @@ public final class PrimaryRegistry {
 
     /**
      * A recorded delegator: the terminal learned from call traffic, its name if it has one, and
-     * whether a nudge may tell it to {@code fleet_poll(target=…)}, {@code fleet_send(turnId=…)},
-     * or {@code fleet_poll(ticket=…)} — the real authorization decision for this delegator's
-     * role, captured once by the caller at delegation time rather than re-derived from a bare
-     * terminal string later.
+     * whether a nudge may tell it to {@code fleet_poll(target=…)} or {@code fleet_send(turnId=…)}
+     * — the real authorization decision for this delegator's role, captured once by the caller at
+     * delegation time rather than re-derived from a bare terminal string later.
      */
-    private record Delegation(String terminal, String name, boolean mayDrainNudge, boolean mayAnswerNudge,
-                              boolean mayTaskReadNudge) {
+    private record Delegation(String terminal, String name, boolean mayDrainNudge, boolean mayAnswerNudge) {
     }
 
     /**
@@ -137,11 +135,11 @@ public final class PrimaryRegistry {
      * As {@link #recordDelegation(String, String)}, additionally recording the delegating lead's
      * name when the caller carries one, and granting a nudge about this target everything a
      * primary may be told to run. See {@link #record(String, String)} for why the name matters,
-     * and {@link #recordDelegation(String, String, String, boolean, boolean, boolean)} for why
-     * every other caller must supply its own, real grant instead of this default.
+     * and {@link #recordDelegation(String, String, String, boolean, boolean)} for why every other
+     * caller must supply its own, real grant instead of this default.
      */
     public void recordDelegation(String target, String leadTerminal, String leadName) {
-        recordDelegation(target, leadTerminal, leadName, true, true, true);
+        recordDelegation(target, leadTerminal, leadName, true, true);
     }
 
     /**
@@ -150,28 +148,13 @@ public final class PrimaryRegistry {
      * role, asked once by the caller at delegation time, since a terminal string alone cannot be
      * resolved back to a role here. A nudge about this target must offer only what
      * {@link #nudgeMayDrainFor(String)}/{@link #nudgeMayAnswerFor(String)} report back as true.
-     * {@code TASK_READ} defaults to granted — see
-     * {@link #recordDelegation(String, String, String, boolean, boolean, boolean)} for the overload
-     * that supplies its own grant instead.
      */
     public void recordDelegation(String target, String leadTerminal, String leadName,
                                   boolean mayDrainNudge, boolean mayAnswerNudge) {
-        recordDelegation(target, leadTerminal, leadName, mayDrainNudge, mayAnswerNudge, true);
-    }
-
-    /**
-     * As {@link #recordDelegation(String, String, String, boolean, boolean)}, also with the
-     * delegating caller's own {@code TASK_READ} grant for a ticket it created — asked once at
-     * delegation time, the same way, for a nudge to offer only what
-     * {@link #nudgeMayTaskReadFor(String)} reports back as true.
-     */
-    public void recordDelegation(String target, String leadTerminal, String leadName,
-                                  boolean mayDrainNudge, boolean mayAnswerNudge, boolean mayTaskReadNudge) {
         if (target == null || target.isBlank() || leadTerminal == null || leadTerminal.isBlank()) {
             return;
         }
-        leadByTarget.put(target,
-                new Delegation(leadTerminal, blankToNull(leadName), mayDrainNudge, mayAnswerNudge, mayTaskReadNudge));
+        leadByTarget.put(target, new Delegation(leadTerminal, blankToNull(leadName), mayDrainNudge, mayAnswerNudge));
     }
 
     /** Forget a worker's delegating lead — call on release, so a torn-down session leaks nothing. */
@@ -229,18 +212,6 @@ public final class PrimaryRegistry {
         return currentPrimaryTerminal().isPresent() ? Optional.of(true) : Optional.empty();
     }
 
-    /**
-     * As {@link #nudgeMayDrainFor(String)}, for whether a nudge may tell its recipient to
-     * {@code fleet_poll(ticket=…)}.
-     */
-    public Optional<Boolean> nudgeMayTaskReadFor(String target) {
-        Delegation delegation = target == null ? null : leadByTarget.get(target);
-        if (delegation != null) {
-            return Optional.of(delegation.mayTaskReadNudge());
-        }
-        return currentPrimaryTerminal().isPresent() ? Optional.of(true) : Optional.empty();
-    }
-
     /**
      * The known primary terminal, or empty if not yet learned (and not pinned) — the raw value as
      * it was recorded, with no attempt to resolve a named lead's current pane. Callers that need a

fleetd/src/main/java/dev/ltms/fleet/msg/ReplyPushLoop.java

@@ -195,23 +195,15 @@ public final class ReplyPushLoop {
     }
 
     /**
-     * A ticket awaiting collection: which lead to nudge, whether it ended in failure, how many
-     * nudges have named it so far (CB-598 — tracked per ticket, not per lead per source), and
-     * whether that nudge target's own role may actually run the {@code fleet_poll(ticket=…)} a
-     * ticket nudge would tell it to run.
+     * A ticket awaiting collection: which lead to nudge, whether it ended in failure, and how
+     * many nudges have named it so far (CB-598 — tracked per ticket, not per lead per source).
      */
-    private record PendingTicket(String ticket, String lead, boolean failed, int nudgeCount,
-                                  boolean mayTaskReadNudge) {
+    private record PendingTicket(String ticket, String lead, boolean failed, int nudgeCount) {
     }
 
     /** Tickets still pending for {@code lead}, snapshotted fresh for one tick. */
     private List<PendingTicket> pendingTicketsFor(String lead) {
-        // Never nudge a caller to run a TASK_READ its own role could never perform -- the durable
-        // inbox / the caller's own later poll stays the backstop for it, same as when no nudge
-        // target is known at all.
-        return pendingTickets.values().stream()
-                .filter(t -> lead.equals(t.lead()) && t.mayTaskReadNudge())
-                .toList();
+        return pendingTickets.values().stream().filter(t -> lead.equals(t.lead())).toList();
     }
 
     /** Ticket ids still pending for {@code lead} — a plain snapshot for race comparison. */
@@ -570,10 +562,8 @@ public final class ReplyPushLoop {
                     ticket, target);
             return;
         }
-        boolean mayTaskReadNudge = primaryRegistry.nudgeMayTaskReadFor(target).orElse(false);
         pendingTickets.compute(ticket, (id, existing) ->
-                new PendingTicket(ticket, lead.get(), failed, existing == null ? 0 : existing.nudgeCount(),
-                        mayTaskReadNudge));
+                new PendingTicket(ticket, lead.get(), failed, existing == null ? 0 : existing.nudgeCount()));
         startOrCoalesce(lead.get());
     }
 
@@ -844,8 +834,7 @@ public final class ReplyPushLoop {
         }
         for (PendingTicket ticket : tickets) {
             pendingTickets.computeIfPresent(ticket.ticket(),
-                    (id, e) -> new PendingTicket(e.ticket(), e.lead(), e.failed(), e.nudgeCount() + 1,
-                            e.mayTaskReadNudge()));
+                    (id, e) -> new PendingTicket(e.ticket(), e.lead(), e.failed(), e.nudgeCount() + 1));
         }
         for (PendingQuestion question : questions) {
             pendingQuestions.computeIfPresent(question.turnId(), (id, e) ->

fleetd/src/test/java/dev/ltms/fleet/msg/ReplyPushLoopTest.java

@@ -398,9 +398,9 @@ class ReplyPushLoopTest {
     }
 
     /**
-     * A delegator barred from DRAIN can still have a ticket genuinely pending for it — the 5-arg
-     * {@code recordDelegation} overload used here defaults {@code mayTaskReadNudge} to granted, so
-     * the ticket half still fires while the reply half must still be suppressed.
+     * A delegator barred from DRAIN can still have a ticket genuinely pending for it — tickets are
+     * out of this unit's scope (fleetd #778 Part 1 already makes a ticket nudge correct for an
+     * observer). The reply half must still be suppressed even though the ticket half fires.
      */
     @Test
     void aForbiddenReplyNudgeIsSuppressedWhileAnEligibleTicketStillFires() throws Exception {
@@ -418,35 +418,6 @@ class ReplyPushLoopTest {
                 "the reply half is forbidden for this delegator and must never be named: " + nudge);
     }
 
-    // --- fleetd #804: never nudge a delegator to run a ticket TASK_READ its own role is refused --
-
-    @Test
-    void onTicketTerminalNeverNudgesADelegatorThatMayNotRunTaskRead() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        registry.recordDelegation(WORKER, OTHER_PRIMARY, "someone", true, true, false);
-
-        loop(1, 50).onTicketTerminal("task-1", WORKER, false);
-
-        Thread.sleep(200);
-        assertEquals(0, rec.sendCount(),
-                "a delegator whose role cannot run fleet_poll(ticket=...) must never be nudged to run it");
-    }
-
-    @Test
-    void onTicketTerminalStillNudgesADelegatorThatMayRunTaskRead() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        registry.recordDelegation(WORKER, OTHER_PRIMARY, "lead", true, true, true);
-
-        loop(1, 50).onTicketTerminal("task-1", WORKER, false);
-
-        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS));
-        String nudge = rec.sentParams().getFirst().getValue().toString();
-        assertTrue(nudge.contains("fleet_poll(ticket=task-1)"),
-                "a delegator that may run TASK_READ must still be nudged about it: " + nudge);
-    }
-
     // --- nudge format --------------------------------------------------------------------------
 
     @Test