refactor: prefer streams for collection transforms; tighten conventions

Convert imperative collection-transform loops to Stream pipelines across
core/runtime/client where the body is pure map/filter/reduce (no checked
exceptions, no primitives, no I/O). Examples: every enum fromWire, the
Lease capability builder, AgentRegistry's resolve + describe, the
SessionLoop job-filter chain, and the ResumeBuffer ring snapshots.

Tighten Java conventions while in the area:
- annotate IdempotencyStore.claim nullable return
- make ArcpClient's sealed Message switch exhaustive (no default branch)
- narrow catch (Exception) to RuntimeException where Agent.run is not the
  boundary; narrow WebSocketTransport close to ExecutionException |
  TimeoutException
- move ReplayingPublisher and StdioTransport off synchronized blocks to
  ReentrantLock, with comments explaining why the lock spans blocking I/O
- promote volatile long lastInboundMillis to AtomicLong for consistency
  with neighbouring counters
- TckProvider extends AutoCloseable so ConformanceSuite can drop its
  hand-rolled finally close

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: nficano

arcp-client/src/main/java/dev/arcp/client/ArcpClient.java

@@ -84,9 +84,9 @@ public final class ArcpClient implements AutoCloseable, Flow.Subscriber<Envelope
     private final Duration ackInterval;
     private final AtomicLong lastSeenSeq = new AtomicLong(-1);
     private final AtomicLong lastAckedSeq = new AtomicLong(-1);
+    private final AtomicLong lastInboundMillis = new AtomicLong(System.currentTimeMillis());
     private @Nullable ScheduledFuture<?> ackTick;
     private @Nullable ScheduledFuture<?> heartbeatWatchdog;
-    private volatile long lastInboundMillis = System.currentTimeMillis();
     private final ConcurrentHashMap<JobId, SubmissionPublisher<EventBody>> liveSubscribers =
             new ConcurrentHashMap<>();
     @SuppressWarnings("unused")
@@ -212,14 +212,14 @@ public void onSubscribe(Flow.Subscription s) {
 
     @Override
     public void onNext(Envelope envelope) {
-        lastInboundMillis = System.currentTimeMillis();
+        lastInboundMillis.set(System.currentTimeMillis());
         Long seq = envelope.eventSeq();
         if (seq != null) {
             lastSeenSeq.updateAndGet(prev -> Math.max(prev, seq));
         }
         try {
             dispatch(envelope);
-        } catch (Exception e) {
+        } catch (RuntimeException e) {
             log.warn("client dispatch error for {}: {}", envelope.type(), e.toString());
         }
     }
@@ -256,7 +256,15 @@ private void dispatch(Envelope envelope) {
             case JobSubscribed ignored -> { /* signal */ }
             case SessionJobs jobs -> handleListResponse(jobs);
             case SessionPing ping -> handlePing(ping);
-            default -> log.debug("client ignored: {}", envelope.type());
+            case SessionPong ignored -> log.debug("client ignored: {}", envelope.type());
+            case SessionAck ignored -> log.debug("client ignored: {}", envelope.type());
+            case SessionHello ignored -> log.debug("client ignored: {}", envelope.type());
+            case SessionBye ignored -> log.debug("client ignored: {}", envelope.type());
+            case SessionListJobs ignored -> log.debug("client ignored: {}", envelope.type());
+            case JobSubmit ignored -> log.debug("client ignored: {}", envelope.type());
+            case JobCancel ignored -> log.debug("client ignored: {}", envelope.type());
+            case JobSubscribe ignored -> log.debug("client ignored: {}", envelope.type());
+            case JobUnsubscribe ignored -> log.debug("client ignored: {}", envelope.type());
         }
     }
 
@@ -305,7 +313,7 @@ private void maybeAck() {
     }
 
     private void watchHeartbeat(long intervalMs) {
-        long elapsed = System.currentTimeMillis() - lastInboundMillis;
+        long elapsed = System.currentTimeMillis() - lastInboundMillis.get();
         if (elapsed > intervalMs * 2) {
             log.info("client observed heartbeat loss; closing session");
             close();
@@ -319,11 +327,7 @@ private void handleAccepted(Envelope envelope, JobAccepted accepted) {
         // We associate by traversing pending submits in insertion order; the
         // runtime guarantees ordering per-session, so the oldest pending submit
         // is the one being acknowledged.
-        MessageId match = null;
-        for (var entry : pendingSubmits.entrySet()) {
-            match = entry.getKey();
-            break;
-        }
+        MessageId match = pendingSubmits.keySet().stream().findFirst().orElse(null);
         if (match == null) {
             return;
         }
@@ -370,11 +374,7 @@ private void handleError(Envelope envelope, JobError err) {
         Outstanding o = jid != null ? outstanding.remove(jid) : null;
         if (o == null) {
             // Top-level (unassigned) error: drop the oldest pending submit.
-            MessageId first = null;
-            for (var k : pendingSubmits.keySet()) {
-                first = k;
-                break;
-            }
+            MessageId first = pendingSubmits.keySet().stream().findFirst().orElse(null);
             if (first != null) {
                 Outstanding pending = pendingSubmits.remove(first);
                 if (pending != null) {

arcp-client/src/main/java/dev/arcp/client/ReplayingPublisher.java

@@ -7,6 +7,7 @@
 import java.util.concurrent.Flow;
 import java.util.concurrent.SubmissionPublisher;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Multicast {@link Flow.Publisher} that buffers every emission and replays the
@@ -18,24 +19,37 @@ final class ReplayingPublisher<T> implements Flow.Publisher<T> {
 
     private final List<T> buffer = new CopyOnWriteArrayList<>();
     private final SubmissionPublisher<T> live;
+    private final ReentrantLock lock = new ReentrantLock();
     private volatile boolean closed;
 
     ReplayingPublisher() {
         this.live = new SubmissionPublisher<>(
                 Executors.newVirtualThreadPerTaskExecutor(), 1024);
     }
 
-    synchronized void submit(T item) {
-        buffer.add(item);
-        live.submit(item);
+    // Lock spans live.submit so concurrent producers preserve buffer/live order;
+    // back-pressure blocking on a full submission queue therefore stalls peers.
+    void submit(T item) {
+        lock.lock();
+        try {
+            buffer.add(item);
+            live.submit(item);
+        } finally {
+            lock.unlock();
+        }
     }
 
-    synchronized void close() {
-        if (closed) {
-            return;
+    void close() {
+        lock.lock();
+        try {
+            if (closed) {
+                return;
+            }
+            closed = true;
+            live.close();
+        } finally {
+            lock.unlock();
         }
-        closed = true;
-        live.close();
     }
 
     @Override
@@ -46,7 +60,8 @@ public void subscribe(Flow.Subscriber<? super T> downstream) {
 
         // Hold the publisher lock while snapshotting AND attaching to live so
         // no submit() can interleave between the two and produce a gap.
-        synchronized (this) {
+        lock.lock();
+        try {
             snapshot = new ArrayList<>(buffer);
             wasClosed = closed;
             if (!wasClosed) {
@@ -78,6 +93,8 @@ public void onComplete() {
                     }
                 });
             }
+        } finally {
+            lock.unlock();
         }
 
         downstream.onSubscribe(new Flow.Subscription() {

arcp-client/src/main/java/dev/arcp/client/WebSocketTransport.java

@@ -18,6 +18,7 @@
 import java.util.concurrent.Flow;
 import java.util.concurrent.SubmissionPublisher;
 import java.util.concurrent.TimeUnit;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,7 +64,7 @@ public void onOpen(WebSocket webSocket) {
             }
 
             @Override
-            public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
+            public @Nullable CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
                 WebSocketTransport t = futureSocket.get();
                 if (t != null) {
                     t.handleText(data, last);
@@ -73,7 +74,7 @@ public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean
             }
 
             @Override
-            public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
+            public @Nullable CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
                 WebSocketTransport t = futureSocket.get();
                 if (t != null) {
                     t.inbound.close();
@@ -146,7 +147,8 @@ public void close() {
             socket.sendClose(WebSocket.NORMAL_CLOSURE, "bye").get(2, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
-        } catch (Exception ignored) {
+        } catch (java.util.concurrent.ExecutionException
+                | java.util.concurrent.TimeoutException ignored) {
             // best-effort close
         }
         inbound.close();

arcp-core/src/main/java/dev/arcp/core/capabilities/Capabilities.java

@@ -3,12 +3,11 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.jspecify.annotations.Nullable;
 
 /**
@@ -35,25 +34,19 @@ static Capabilities fromJson(
             @JsonProperty("encodings") @Nullable List<String> encodings,
             @JsonProperty("features") @Nullable List<String> features,
             @JsonProperty("agents") @Nullable List<AgentDescriptor> agents) {
-        Set<Feature> parsed = EnumSet.noneOf(Feature.class);
-        if (features != null) {
-            for (String w : features) {
-                Feature.fromWire(w).ifPresent(parsed::add);
-            }
-        }
+        Set<Feature> parsed = features == null
+                ? EnumSet.noneOf(Feature.class)
+                : features.stream()
+                        .flatMap(w -> Feature.fromWire(w).stream())
+                        .collect(Collectors.toCollection(() -> EnumSet.noneOf(Feature.class)));
         return new Capabilities(
                 encodings == null ? List.of("json") : encodings, parsed, agents);
     }
 
     @JsonProperty("features")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     public List<String> featuresWire() {
-        List<String> out = new ArrayList<>(features.size());
-        for (Feature f : features) {
-            out.add(f.wire());
-        }
-        Collections.sort(out);
-        return out;
+        return features.stream().map(Feature::wire).sorted().toList();
     }
 
     public static Set<Feature> intersect(Set<Feature> a, Set<Feature> b) {

arcp-core/src/main/java/dev/arcp/core/capabilities/Feature.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonValue;
+import java.util.Arrays;
 import java.util.Optional;
 
 public enum Feature {
@@ -28,11 +29,8 @@ public String wire() {
 
     @JsonCreator
     public static Optional<Feature> fromWire(String wire) {
-        for (Feature f : values()) {
-            if (f.wire.equals(wire)) {
-                return Optional.of(f);
-            }
-        }
-        return Optional.empty();
+        return Arrays.stream(values())
+                .filter(f -> f.wire.equals(wire))
+                .findFirst();
     }
 }

arcp-core/src/main/java/dev/arcp/core/error/ErrorCode.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonValue;
+import java.util.Arrays;
 
 public enum ErrorCode {
     PERMISSION_DENIED(false),
@@ -37,11 +38,9 @@ public String wire() {
 
     @JsonCreator
     public static ErrorCode fromWire(String wire) {
-        for (ErrorCode c : values()) {
-            if (c.name().equals(wire)) {
-                return c;
-            }
-        }
-        return INTERNAL_ERROR;
+        return Arrays.stream(values())
+                .filter(c -> c.name().equals(wire))
+                .findFirst()
+                .orElse(INTERNAL_ERROR);
     }
 }

arcp-core/src/main/java/dev/arcp/core/events/EventBody.java

@@ -42,12 +42,10 @@ public String wire() {
         }
 
         public static Kind fromWire(String wire) {
-            for (Kind k : values()) {
-                if (k.wire.equals(wire)) {
-                    return k;
-                }
-            }
-            throw new IllegalArgumentException("unknown event kind: " + wire);
+            return java.util.Arrays.stream(values())
+                    .filter(k -> k.wire.equals(wire))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalArgumentException("unknown event kind: " + wire));
         }
     }
 }

arcp-core/src/main/java/dev/arcp/core/lease/Lease.java

@@ -11,6 +11,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * Capability bag: namespace → list of pattern strings.
@@ -24,11 +25,12 @@ public final class Lease {
 
     public Lease(Map<String, List<String>> capabilities) {
         Objects.requireNonNull(capabilities, "capabilities");
-        Map<String, List<String>> copy = new LinkedHashMap<>();
-        for (var e : capabilities.entrySet()) {
-            copy.put(e.getKey(), List.copyOf(e.getValue()));
-        }
-        this.capabilities = Collections.unmodifiableMap(copy);
+        this.capabilities = Collections.unmodifiableMap(capabilities.entrySet().stream()
+                .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        e -> List.copyOf(e.getValue()),
+                        (a, b) -> a,
+                        LinkedHashMap::new)));
     }
 
     public static Lease empty() {
@@ -46,17 +48,21 @@ public List<String> patterns(String namespace) {
 
     /** Parsed cost.budget initial amounts per currency. */
     public Map<String, BigDecimal> budget() {
-        Map<String, BigDecimal> out = new LinkedHashMap<>();
-        for (String entry : patterns("cost.budget")) {
-            int colon = entry.indexOf(':');
-            if (colon <= 0 || colon == entry.length() - 1) {
-                throw new IllegalArgumentException("invalid cost.budget pattern: " + entry);
-            }
-            String currency = entry.substring(0, colon);
-            BigDecimal amount = new BigDecimal(entry.substring(colon + 1));
-            out.merge(currency, amount, BigDecimal::add);
-        }
-        return out;
+        return patterns("cost.budget").stream()
+                .map(entry -> {
+                    int colon = entry.indexOf(':');
+                    if (colon <= 0 || colon == entry.length() - 1) {
+                        throw new IllegalArgumentException("invalid cost.budget pattern: " + entry);
+                    }
+                    return Map.entry(
+                            entry.substring(0, colon),
+                            new BigDecimal(entry.substring(colon + 1)));
+                })
+                .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        Map.Entry::getValue,
+                        BigDecimal::add,
+                        LinkedHashMap::new));
     }
 
     @JsonCreator
@@ -66,18 +72,10 @@ static Lease fromJson(Map<String, List<String>> wire) {
 
     /** §9.4 subset check: every key/pattern in {@code child} appears in {@code this}. */
     public boolean contains(Lease child) {
-        for (var e : child.capabilities.entrySet()) {
+        return child.capabilities.entrySet().stream().allMatch(e -> {
             List<String> parent = capabilities.get(e.getKey());
-            if (parent == null) {
-                return false;
-            }
-            for (String pat : e.getValue()) {
-                if (!parent.contains(pat)) {
-                    return false;
-                }
-            }
-        }
-        return true;
+            return parent != null && parent.containsAll(e.getValue());
+        });
     }
 
     public static final class Builder {

arcp-core/src/main/java/dev/arcp/core/messages/Message.java

@@ -56,12 +56,10 @@ public String wire() {
         }
 
         public static Type fromWire(String wire) {
-            for (Type t : values()) {
-                if (t.wire.equals(wire)) {
-                    return t;
-                }
-            }
-            throw new IllegalArgumentException("unknown message type: " + wire);
+            return java.util.Arrays.stream(values())
+                    .filter(t -> t.wire.equals(wire))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalArgumentException("unknown message type: " + wire));
         }
     }
 }