fix(sdk-java): close hosted harness ledger and SSE close races

- Classify a parked SSE read's raced close as DaemonTransportException, matching the interrupted-read signal, with the idle-timeout half kept distinct; the programming-error type stays reserved for callers arriving after the close. - Keep the admission ledger on every refused close, on a settled release whose entry is releaseBlocked, and for a code-less unfenced 409, which is outcome-unknown rather than a definitive refusal. - Retire the marker an owning same-identity replay just planted when the promptId's terminal event was already observed, via the settled-checked release path. - Read the getStatus marker clock before settledness and re-check both inside the bin-locked clear, so a retry's beginSend landing between the snapshot reads cannot be cleared under its open send window. - Retire only the observed attachment on a by-id close, and self-retire an attachment whose heartbeat meets a definitive hosted_session_not_found 404 instead of beating against the dead session forever.
2026-11-07 06:19:43 -10:00 · 2026-10-14 00:36:31 +00:00
1 parent 4c4a9e090c
commit 4884dc8197
@@ -57,6 +57,13 @@ public final class DaemonEvent {
        if (!("turn_complete".equals(type) || "turn_error".equals(type))) {
            return expectedPromptId.equals(promptId);
        }
+        return expectedPromptId.equals(terminalPromptId());
+    }
+
+    // The promptId a terminal event settles: the envelope field when
+    // present, else the data object's — null when the event carries
+    // neither.
+    String terminalPromptId() {
        Map<String, Object> object = dataObject();
        String dataPromptId = object == null ? null
                : JsonSupport.optionalString(object, "promptId");
@@ -65,8 +72,7 @@ public final class DaemonEvent {
            throw new DaemonProtocolException(
                    "SSE envelope promptId conflicts with data.promptId");
        }
-        String canonicalPromptId = promptId == null ? dataPromptId : promptId;
-        return expectedPromptId.equals(canonicalPromptId);
+        return promptId == null ? dataPromptId : promptId;
    }

    String updateKind() {
@@ -73,9 +73,20 @@ public final class HarnessEventStream implements AutoCloseable {
            try {
                SseReader.Frame frame = reader.next();
                // A close() that raced this read (a buffered frame, or the
-                // end of the body) reports as closed, not as an event or
-                // as the server ending the stream.
-                ensureOpen();
+                // end of the body) reports the same signal the interrupted
+                // read would have raised — one cause, one exception type
+                // however it raced — never as an event or the server
+                // ending the stream. The closed guard's programming-error
+                // type stays reserved for a caller that arrives after the
+                // close (the pre-park guards above).
+                if (closed.get()) {
+                    if (idleTimedOut.get()) {
+                        throw new DaemonTransportException(
+                                "Hosted Harness SSE idle timeout");
+                    }
+                    throw new DaemonTransportException(
+                            "HarnessEventStream is closed");
+                }
                if (frame == null) {
                    close();
                    return null;
@@ -207,11 +218,4 @@ public final class HarnessEventStream implements AutoCloseable {
                    e);
        }
    }
-
-    private void ensureOpen() {
-        if (closed.get()) {
-            throw new IllegalStateException(
-                    "HarnessEventStream is closed");
-        }
-    }
 }
@@ -82,6 +82,16 @@ public final class HostedHarnessClient implements AutoCloseable {
            new ConcurrentHashMap<>();
    private final Map<String, ActivePrompt> activePrompts =
            new ConcurrentHashMap<>();
+    // The promptId of each session's last observed terminal event: a
+    // replay 202 answered from the route's journal for such a promptId owes
+    // no future terminal event, so an owning retry must not keep the marker
+    // it just planted for it. Retired with the rest of a session
+    // incarnation's local state.
+    private final Map<String, String> terminalPromptIds =
+            new ConcurrentHashMap<>();
+    // Test seam: parks a getStatus snapshot between its two reads,
+    // modelling an OS preemption at that point.
+    volatile Runnable statusSnapshotProbe = () -> {};
    private final Set<HarnessEventStream> streams =
            ConcurrentHashMap.newKeySet();
    private final AtomicBoolean closed = new AtomicBoolean();
@@ -383,7 +393,10 @@ public final class HostedHarnessClient implements AutoCloseable {
                // path whose unfenced 4xx still surfaces as DaemonHttpException:
                // the marker release keys on the refusal's definitiveness, and
                // the turn consumer's terminal classification of a refused
-                // submission is bounded (hosted_harness_rejected).
+                // submission is bounded (hosted_harness_rejected). A 409 is the
+                // exception: the consumer waits coded ones out as
+                // duplicate-admission signals, so a code-less one is
+                // outcome-unknown below, never this surface.
                if (DaemonClient.isAmbiguousMutationStatus(raw.statusCode())
                        || raw.statusCode() < 400) {
                    if (!ownsActivePrompt) {
@@ -409,6 +422,19 @@ public final class HostedHarnessClient implements AutoCloseable {
                                "POST /session/:id/prompt", raw.statusCode(),
                                refusalBody, code);
                    }
+                    if (code == null) {
+                        // A boot-headerless 409 whose body carries no
+                        // refusal-vocabulary code is not the route's answer
+                        // (an intermediary's error page, a body truncated
+                        // past the buffered bound): like the unfenced
+                        // 408/5xx arm above it cannot prove this submission
+                        // was refused, so it is outcome-unknown and the
+                        // marker stays.
+                        if (!ownsActivePrompt) {
+                            registered.blockRelease();
+                        }
+                        throw new PromptAdmissionUnknownException(e);
+                    }
                }
                if (ownsActivePrompt) {
                    releaseOwnedMarker(session.getHarnessSessionId(),
@@ -522,6 +548,17 @@ public final class HostedHarnessClient implements AutoCloseable {
                    // the finally below, after this republish.
                    activePrompts.replace(session.getHarnessSessionId(),
                            registered, candidate);
+                } else if (candidate.promptId.equals(terminalPromptIds.get(
+                        session.getHarnessSessionId()))) {
+                    // A same-identity retry answered from the journal after
+                    // this promptId's terminal event was already observed
+                    // owes no future terminal event, so the marker this
+                    // call just planted could never be cleared: release it
+                    // through the settled-checked owning path — a third
+                    // same-identity attempt may have joined the entry. A
+                    // fresh admission keeps the marker its turn still owes.
+                    releaseOwnedMarker(session.getHarnessSessionId(),
+                            candidate);
                }
                return new PromptReceipt(responsePromptId,
                        JsonSupport.requiredNonNegativeLong(json, "lastEventId",
@@ -917,8 +954,13 @@ public final class HostedHarnessClient implements AutoCloseable {
        // says nothing about it.
        ActivePrompt observed = activePrompts.get(
                ref.getHarnessSessionId());
-        boolean observedSettled = observed != null && observed.isSettled();
+        // The clock is read before settledness: a beginSend landing between
+        // the two snapshot reads then flips the settled read false, and the
+        // clear below re-checks settledness inside the bin-locked compute,
+        // where the retry's beginSend cannot interleave.
+        statusSnapshotProbe.run();
        long observedClock = observed != null ? observed.sendClock() : 0;
+        boolean observedSettled = observed != null && observed.isSettled();
        String operation = "GET /session/:id/status";
        HttpSupport.Response response = sendRead(
                sessionPath(ref.getHarnessSessionId()) + "/status", ref,
@@ -942,7 +984,8 @@ public final class HostedHarnessClient implements AutoCloseable {
            // concluded during the round trip may postdate it.
            activePrompts.compute(ref.getHarnessSessionId(),
                    (sessionId, current) -> current == observed
-                            && observed.sendClock() == observedClock
+                            && current.isSettled()
+                            && current.sendClock() == observedClock
                            ? null : current);
        }
        return new HarnessSessionStatus(responseSessionId, active, json);
@@ -1155,11 +1198,6 @@ public final class HostedHarnessClient implements AutoCloseable {

    public void closeSession(HarnessSessionRef session) {
        HarnessSessionRef ref = requireSessionRef(session);
-        // Snapshot the ledger entry before the wire call: a refusal arm
-        // may retire only the record it observed, or an answer landing
-        // after a concurrent re-registration discards state it never saw.
-        ActivePrompt observedLedger = activePrompts.get(
-                ref.getHarnessSessionId());
        try {
            boolean deleted = closeSession(ref.getHarnessSessionId(),
                    ref.getHarnessClientId());
@@ -1180,22 +1218,18 @@ public final class HostedHarnessClient implements AutoCloseable {
                        ref.getHarnessClientId());
            }
        } catch (DaemonHttpException | HostedHarnessGenerationException e) {
-            // A definitive refusal still retires the local state; an
-            // outcome-unknown surface keeps it, same rule as detach. The
-            // ledger is the one exception: a refusal proving the turn is
-            // still live (hosted_turn_active) must keep the admission
-            // record, or the local one-turn veto fails open against a
-            // running turn.
+            // A definitive refusal proves the DELETE did not run — the
+            // session and any turn on it are still alive — so it retires
+            // the attachment but keeps the admission ledger: a refusal
+            // speaks only to the DELETE, never to a submission still in
+            // flight, and the kept marker is the one record whose veto
+            // fails closed. Same rule as detachSession above. A proven
+            // generation change instead means every local record for that
+            // peer is dead, including a marker published while the DELETE
+            // was in flight.
            removeAttachment(ref);
            if (e instanceof HostedHarnessGenerationException) {
-                // A proven generation change means every local record for
-                // that peer is dead — including a marker published while
-                // the DELETE was in flight, which the observed snapshot
-                // cannot see. Same rule as detachSession above.
                activePrompts.remove(ref.getHarnessSessionId());
-            } else if (!provesLiveTurn(e)) {
-                removeObservedLedger(ref.getHarnessSessionId(),
-                        observedLedger);
            }
            throw e;
        }
@@ -1205,23 +1239,23 @@ public final class HostedHarnessClient implements AutoCloseable {
        ensureOpen();
        String sessionId = requireUuid(harnessSessionId,
                "harnessSessionId");
-        // Snapshot both records before the DELETE: a refusal that lands
+        // Snapshot the attachment before the DELETE: an arm that lands
        // after a concurrent loadSession may retire only what it observed,
-        // or it wipes the newer caller's attachment and admission marker —
-        // AttachmentState.matches cannot key this path because it compares
-        // client ids and this overload holds none. A success deleted the
-        // session outright, so the success arm's retirement below is
-        // unconditional: every attachment and marker for the id is stale,
-        // including one registered while the DELETE was in flight.
+        // or it wipes the newer caller's attachment — AttachmentState.matches
+        // cannot key this path because it compares client ids and this
+        // overload holds none. The 204 answers a DELETE dispatched before
+        // any such re-registration, so a survivor may belong to a re-opened
+        // incarnation whose heartbeats the peer still accepts; when it does
+        // not, its heartbeat's definitive hosted_session_not_found 404
+        // retires it (see scheduleHeartbeat), capping a stale task at one
+        // heartbeat interval. The ledger retirement on success stays
+        // unconditional: the headerless DELETE is answered with
+        // allowMissingClientId, so its success deleted the session outright
+        // and every marker for the id is stale regardless of identity.
        AttachmentState observedAttachment = attachments.get(sessionId);
-        ActivePrompt observedLedger = activePrompts.get(sessionId);
        try {
            closeSession(sessionId, null);
-            discardAnyAttachment(sessionId);
-            // The headerless DELETE this overload sends is answered with
-            // allowMissingClientId, so its success deleted the session
-            // outright: every marker for it is stale regardless of the
-            // submitting identity.
+            discardObservedAttachment(sessionId, observedAttachment);
            retireLedger(sessionId, null);
        } catch (DaemonHttpException | HostedHarnessGenerationException e) {
            if (e instanceof HostedHarnessGenerationException) {
@@ -1230,25 +1264,20 @@ public final class HostedHarnessClient implements AutoCloseable {
                // registered while the DELETE was in flight.
                discardLocalSessionState(sessionId);
            } else {
-                discardLocalSessionState(sessionId, observedAttachment,
-                        observedLedger, provesLiveTurn(e));
+                // Same refusal rule as the ref overload: the refusal proves
+                // the DELETE did not run, so only the observed attachment
+                // retires; the ledger stays as the record of a turn that
+                // may still be live.
+                discardObservedAttachment(sessionId, observedAttachment);
            }
            throw e;
        }
    }

    private void discardLocalSessionState(String sessionId) {
-        discardLocalSessionState(sessionId, attachments.get(sessionId),
-                activePrompts.get(sessionId), false);
-    }
-
-    private void discardLocalSessionState(String sessionId,
-            AttachmentState observedAttachment, ActivePrompt observedLedger,
-            boolean keepLedger) {
-        discardObservedAttachment(sessionId, observedAttachment);
-        if (!keepLedger) {
-            removeObservedLedger(sessionId, observedLedger);
-        }
+        discardObservedAttachment(sessionId, attachments.get(sessionId));
+        removeObservedLedger(sessionId, activePrompts.get(sessionId));
+        terminalPromptIds.remove(sessionId);
    }

    private void discardObservedAttachment(String sessionId,
@@ -1259,17 +1288,6 @@ public final class HostedHarnessClient implements AutoCloseable {
        }
    }

-    // A close that deleted the session retires every attachment for it,
-    // including one registered while the DELETE was in flight: its
-    // heartbeat can never succeed again, and the swallowed 404 would keep
-    // the periodic task on the shared scheduler forever.
-    private void discardAnyAttachment(String sessionId) {
-        AttachmentState state = attachments.remove(sessionId);
-        if (state != null) {
-            state.cancel();
-        }
-    }
-
    // Retire the admission ledger entry a lifecycle teardown proved stale.
    // A caller holding an identity retires only a marker that identity
    // planted: the Harness answers an absent session and a client-id
@@ -1280,6 +1298,11 @@ public final class HostedHarnessClient implements AutoCloseable {
    // so every marker for it is stale. The two-arg remove still leaves a
    // marker a concurrent re-admission published after the read.
    private void retireLedger(String sessionId, String clientId) {
+        if (clientId == null) {
+            // A deletion proof retires the terminal-event record with the
+            // ledger: it spoke for the dead incarnation.
+            terminalPromptIds.remove(sessionId);
+        }
        ActivePrompt marker = activePrompts.get(sessionId);
        if (marker != null
                && (clientId == null || clientId.equals(marker.clientId))) {
@@ -1309,7 +1332,12 @@ public final class HostedHarnessClient implements AutoCloseable {
                return active;
            }
            if (active.isSettled()) {
-                return null;
+                // Honour the same flag dischargeOwedRelease does: a
+                // conclusion that was outcome-unknown or durably admitted
+                // blocks the release — this promptId may be admitted
+                // server-side, so the marker keeps vetoing until a status
+                // read (settled and clock-matched) clears it.
+                return active.releaseBlocked ? active : null;
            }
            // A sibling attempt still holds the entry open: record the owed
            // release for the last concluded attempt to discharge instead of
@@ -1409,8 +1437,18 @@ public final class HostedHarnessClient implements AutoCloseable {
        // ended is the one the ledger tracks regardless of which object
        // currently represents it.
        activePrompts.compute(session.getHarnessSessionId(),
-                (sessionId, active) -> active != null
-                        && event.belongsTo(active.promptId) ? null : active);
+                (sessionId, active) -> {
+                    // Written under the same bin lock as the clear: an
+                    // owning same-identity replay answered from the journal
+                    // after this terminal event owes no future terminal
+                    // event, and its 202 arm retires the marker it planted.
+                    String terminalPromptId = event.terminalPromptId();
+                    if (terminalPromptId != null) {
+                        terminalPromptIds.put(sessionId, terminalPromptId);
+                    }
+                    return active != null && event.belongsTo(active.promptId)
+                            ? null : active;
+                });
    }

    void registerStream(HarnessEventStream stream) {
@@ -1812,14 +1850,6 @@ public final class HostedHarnessClient implements AutoCloseable {
                : null;
    }

-    // A refusal whose code proves the session's turn is still running
-    // server-side: the admission ledger must then survive a failed close.
-    private static boolean provesLiveTurn(DaemonException e) {
-        return e instanceof DaemonHttpException
-                && "hosted_turn_active".equals(refusalCode(
-                        ((DaemonHttpException) e).getResponseBody()));
-    }
-
    // A load that fails with a refusal code on the wire is fail-closed, not
    // ambiguous: the Harness answers with a code only after the failed open
    // was cleaned up. Transport failures and code-less error bodies keep the
@@ -2012,8 +2042,25 @@ public final class HostedHarnessClient implements AutoCloseable {
                    // session for the rest of this client's life.
                    discardLocalSessionState(
                            state.session.getHarnessSessionId());
-                } catch (DaemonException ignored) {
-                    // The next scheduled heartbeat is a new keepalive.
+                } catch (DaemonException e) {
+                    // A definitive hosted_session_not_found 404 proves the
+                    // session record is gone server-side — the same proof a
+                    // deleting close or detach rides — so the attachment
+                    // retires itself instead of beating against the dead
+                    // session for the client's lifetime. Any other refusal
+                    // is transient to the keepalive: the next scheduled
+                    // heartbeat is a new attempt.
+                    if (e instanceof DaemonHttpException
+                            && ((DaemonHttpException) e).getStatusCode()
+                                    == 404
+                            && "hosted_session_not_found".equals(
+                                    ((DaemonHttpException) e)
+                                            .getErrorCode())
+                            && attachments.remove(
+                                    state.session.getHarnessSessionId(),
+                                    state)) {
+                        state.cancel();
+                    }
                } finally {
                    state.heartbeatInFlight.set(false);
                }
@@ -10,6 +10,8 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -64,7 +66,8 @@ class HarnessEventStreamCloseTest {

    // A close that lands while next() already holds a whole frame: the
    // frame is dropped rather than delivered off a closed stream, so the
-    // read position does not move.
+    // read position does not move — and the parked consumer gets the same
+    // transport signal as a read the close interrupted on the socket.
    @Test
    @Timeout(60)
    void aCloseRacingABufferedFrameReportsTheStreamClosed()
@@ -72,8 +75,9 @@ class HarnessEventStreamCloseTest {
        assertCloseRacingTheReadReportsClosed(FRAME);
    }

-    // A close that lands as the body ends: the reader sees the close, not
-    // the server ending the stream, which a consumer reads as an early end.
+    // A close that lands as the body ends: the reader sees the close with
+    // the same transport signal, not the server ending the stream, which a
+    // consumer reads as an early end.
    @Test
    @Timeout(60)
    void aCloseRacingTheEndOfTheBodyReportsTheStreamClosed()
@@ -81,6 +85,39 @@ class HarnessEventStreamCloseTest {
        assertCloseRacingTheReadReportsClosed("");
    }

+    // A frame the reader pulls in the tick the idle watchdog commits to
+    // close is discarded with the idle-timeout signal the interrupted read
+    // would have raised — not the programming-error type — so the
+    // consumer's idle-reconnect branch still fires.
+    @Test
+    @Timeout(60)
+    void anIdleAbortRacingABufferedFrameReportsTheIdleTimeout()
+            throws Exception {
+        HostedHarnessClient client = HostedHarnessClient.builder()
+                .baseUri(harness.baseUri())
+                .bearerToken("harness-token")
+                .capabilityDigest(HeldStreamHarness.DIGEST)
+                .sseIdleTimeout(Duration.ofMillis(200))
+                .build();
+        try {
+            HarnessSessionRef session = client.createSession(
+                    CreateHarnessSession.builder()
+                            .harnessSessionId(HeldStreamHarness.SESSION_ID)
+                            .build());
+            HarnessEventStream stream = new HarnessEventStream(client,
+                    session, new DeliverOnClose(FRAME), 1 << 20, 0,
+                    HeldStreamHarness.EVENT_EPOCH);
+            stream.startIdleWatchdog();
+            DaemonTransportException idle = assertThrows(
+                    DaemonTransportException.class, stream::next);
+            assertEquals("Hosted Harness SSE idle timeout",
+                    idle.getMessage());
+            assertEquals(0, stream.getLastEventId());
+        } finally {
+            client.close();
+        }
+    }
+
    private void assertCloseRacingTheReadReportsClosed(String body)
            throws Exception {
        HostedHarnessClient client = harness.client();
@@ -95,8 +132,12 @@ class HarnessEventStreamCloseTest {
                    session, new CloseOnFirstRead(body, opened), 1 << 20, 0,
                    HeldStreamHarness.EVENT_EPOCH);
            opened.set(stream);
-            IllegalStateException closed = assertThrows(
-                    IllegalStateException.class, stream::next);
+            // The close raced the read, so the parked consumer gets the
+            // same transport signal the interrupted read would have raised
+            // — never the closed guard's programming-error type, which
+            // stays reserved for a caller arriving after the close.
+            DaemonTransportException closed = assertThrows(
+                    DaemonTransportException.class, stream::next);
            assertEquals("HarnessEventStream is closed", closed.getMessage());
            assertEquals(0, stream.getLastEventId());
        } finally {
@@ -218,4 +259,46 @@ class HarnessEventStreamCloseTest {
            }
        }
    }
+
+    // A body whose first read blocks until the stream is closed — the idle
+    // watchdog's closeQuietly — and then delivers its bytes: the shape of a
+    // frame landing in the reader's buffer inside the abort tick.
+    private static final class DeliverOnClose extends InputStream {
+        private final CountDownLatch closedLatch = new CountDownLatch(1);
+        private final InputStream bytes;
+
+        DeliverOnClose(String body) {
+            this.bytes = new ByteArrayInputStream(
+                    body.getBytes(StandardCharsets.UTF_8));
+        }
+
+        @Override
+        public int read() throws IOException {
+            awaitClosed();
+            return bytes.read();
+        }
+
+        @Override
+        public int read(byte[] buffer, int offset, int length)
+                throws IOException {
+            awaitClosed();
+            return bytes.read(buffer, offset, length);
+        }
+
+        @Override
+        public void close() {
+            closedLatch.countDown();
+        }
+
+        private void awaitClosed() throws IOException {
+            try {
+                if (!closedLatch.await(30, TimeUnit.SECONDS)) {
+                    throw new IOException("the stream was never closed");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IOException(e);
+            }
+        }
+    }
 }
@@ -3288,8 +3288,10 @@ class HostedHarnessClientTest {
    @Test
    void aRefusedCloseStillTearsDownTheHeartbeat() throws Exception {
        // Mirror of the detach teardown rule on the close path: a
-        // definitive refusal (DELETE 403) still retires the attachment
-        // and the ledger.
+        // definitive refusal (DELETE 403) still retires the attachment,
+        // but it proves the DELETE did not run — the session and its turn
+        // are still alive — so the admission ledger stays and keeps
+        // vetoing a different identity locally.
        createSessionRoute();
        AtomicInteger heartbeats = new AtomicInteger();
        server.createContext("/session/" + SESSION_ID + "/heartbeat",
@@ -3304,9 +3306,16 @@ class HostedHarnessClientTest {
        server.createContext("/session/" + SESSION_ID + "/prompt",
                exchange -> {
                    int call = promptCalls.incrementAndGet();
+                    if (call > 1) {
+                        // Reached only when the refused close dropped the
+                        // ledger: the production route answers a second
+                        // admission on a live turn with this refusal.
+                        sendSessionJson(exchange, 409,
+                                "{\"code\":\"hosted_turn_active\"}");
+                        return;
+                    }
                    sendSessionJson(exchange, 202,
-                            "{\"promptId\":\"" + (call == 1 ? PROMPT_ID
-                                    : SECOND_PROMPT_ID)
+                            "{\"promptId\":\"" + PROMPT_ID
                                    + "\",\"lastEventId\":0,"
                                    + "\"eventEpoch\":\""
                                    + EVENT_EPOCH + "\"}");
@@ -3341,19 +3350,20 @@ class HostedHarnessClientTest {
            int first = heartbeats.get();
            Thread.sleep(60);
            assertEquals(first, heartbeats.get());
-            // The ledger retired with the attachment: a different identity
-            // is not vetoed locally and reaches the wire.
-            PromptReceipt receipt = client.submitTurn(
-                    SubmitHarnessTurn.builder()
+            // The ledger survived the refused close: a different
+            // identity is vetoed locally and never reaches the wire.
+            DaemonException veto = assertThrows(DaemonException.class,
+                    () -> client.submitTurn(SubmitHarnessTurn.builder()
                            .session(session)
                            .promptId(SECOND_PROMPT_ID)
                            .addContent(block)
                            .payloadDigest(
                                    SubmitHarnessTurn.computePayloadDigest(
                                            List.of(block)))
-                            .build());
-            assertEquals(SECOND_PROMPT_ID, receipt.getPromptId());
-            assertEquals(2, promptCalls.get());
+                            .build()));
+            assertEquals("Hosted Harness session already has a running turn",
+                    veto.getMessage());
+            assertEquals(1, promptCalls.get());
        } finally {
            client.close();
        }
@@ -3361,7 +3371,8 @@ class HostedHarnessClientTest {

    @Test
    void aRefusedCloseByIdStillTearsDownTheHeartbeat() throws Exception {
-        // Same teardown rule for the by-id overload.
+        // Same teardown rule for the by-id overload: the attachment
+        // retires, the admission ledger stays.
        createSessionRoute();
        AtomicInteger heartbeats = new AtomicInteger();
        server.createContext("/session/" + SESSION_ID + "/heartbeat",
@@ -3376,9 +3387,15 @@ class HostedHarnessClientTest {
        server.createContext("/session/" + SESSION_ID + "/prompt",
                exchange -> {
                    int call = promptCalls.incrementAndGet();
+                    if (call > 1) {
+                        // Same wire refusal the production route answers a
+                        // second admission on a live turn with.
+                        sendControlPlaneSessionJson(exchange, 409,
+                                "{\"code\":\"hosted_turn_active\"}");
+                        return;
+                    }
                    sendSessionJson(exchange, 202,
-                            "{\"promptId\":\"" + (call == 1 ? PROMPT_ID
-                                    : SECOND_PROMPT_ID)
+                            "{\"promptId\":\"" + PROMPT_ID
                                    + "\",\"lastEventId\":0,"
                                    + "\"eventEpoch\":\""
                                    + EVENT_EPOCH + "\"}");
@@ -3413,18 +3430,21 @@ class HostedHarnessClientTest {
            int first = heartbeats.get();
            Thread.sleep(60);
            assertEquals(first, heartbeats.get());
-            // The ledger retired with the attachment here too.
-            PromptReceipt receipt = client.submitTurn(
-                    SubmitHarnessTurn.builder()
+            // The ledger survived the refused close here too: a
+            // different identity is vetoed locally and never reaches the
+            // wire.
+            DaemonException veto = assertThrows(DaemonException.class,
+                    () -> client.submitTurn(SubmitHarnessTurn.builder()
                            .session(session)
                            .promptId(SECOND_PROMPT_ID)
                            .addContent(block)
                            .payloadDigest(
                                    SubmitHarnessTurn.computePayloadDigest(
                                            List.of(block)))
-                            .build());
-            assertEquals(SECOND_PROMPT_ID, receipt.getPromptId());
-            assertEquals(2, promptCalls.get());
+                            .build()));
+            assertEquals("Hosted Harness session already has a running turn",
+                    veto.getMessage());
+            assertEquals(1, promptCalls.get());
        } finally {
            client.close();
        }
@@ -4438,6 +4458,17 @@ class HostedHarnessClientTest {
        assertTrue(heartbeats.get() > 0, "heartbeat never fired");
    }

+    private static void awaitSchedulerDrained(HostedHarnessClient client)
+            throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (!client.scheduler().getQueue().isEmpty()
+                && System.nanoTime() < deadline) {
+            Thread.sleep(10);
+        }
+        assertTrue(client.scheduler().getQueue().isEmpty(),
+                "the scheduler queue never drained");
+    }
+
    @Test
    void oversizedLoadRefusalCodeStaysOutcomeUnknown() {
        // The refusal code flows into the turn's error_code column, a
@@ -4939,6 +4970,86 @@ class HostedHarnessClientTest {
        assertEquals(3, promptCalls.get());
    }

+    @Test
+    void anOwningReplayedAdmissionAfterTerminalEventPlantsNoMarker()
+            throws Exception {
+        // R17-2: the owning ordering of
+        // aReplayedAdmissionAfterTerminalEventPlantsNoMarker — the turn's
+        // terminal event is consumed BEFORE the same-identity retry, so
+        // the ledger is empty and the retry OWNS the marker it plants. The
+        // route answers the replay from its journal with a bare 202 and
+        // emits no event, so that marker could never be cleared: the 202
+        // arm retires it instead, and a different identity reaches the
+        // wire.
+        createSessionRoute();
+        AtomicInteger promptCalls = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/prompt",
+                exchange -> {
+                    int call = promptCalls.incrementAndGet();
+                    String promptId = call == 3 ? SECOND_PROMPT_ID
+                            : PROMPT_ID;
+                    sendSessionJson(exchange, 202,
+                            "{\"promptId\":\"" + promptId
+                                    + "\",\"lastEventId\":0,"
+                                    + "\"eventEpoch\":\""
+                                    + EVENT_EPOCH + "\"}");
+                });
+        server.createContext("/session/" + SESSION_ID + "/events",
+                exchange -> {
+                    sendSseHeaders(exchange);
+                    exchange.getResponseBody().write(terminalEvent(1,
+                            PROMPT_ID).getBytes(StandardCharsets.UTF_8));
+                    exchange.getResponseBody().flush();
+                    exchange.close();
+                });
+        Map<String, Object> block = Map.of(
+                "type", "text", "text", "owning-replay-after-terminal");
+
+        try (HostedHarnessClient client = newClient()) {
+            HarnessSessionRef session = createSession(client);
+            SubmitHarnessTurn first = SubmitHarnessTurn.builder()
+                    .session(session)
+                    .promptId(PROMPT_ID)
+                    .addContent(block)
+                    .payloadDigest(SubmitHarnessTurn.computePayloadDigest(
+                            List.of(block)))
+                    .build();
+            assertNotNull(client.submitTurn(first));
+            HarnessEventStream stream = client.streamEvents(
+                    StreamHarnessEvents.builder()
+                            .session(session)
+                            .eventEpoch(EVENT_EPOCH)
+                            .build());
+            try {
+                // The terminal event is consumed before the retry:
+                // observeEvent clears the marker and records the terminal
+                // promptId.
+                DaemonEvent ended = assertTimeoutPreemptively(
+                        Duration.ofSeconds(5), stream::next);
+                assertEquals("turn_complete", ended.getType());
+            } finally {
+                stream.closeQuietly();
+            }
+            // The owning replay is answered from the journal with a bare
+            // 202: the marker it just planted owes no future terminal
+            // event and must not survive.
+            assertNotNull(client.submitTurn(first));
+            // A different identity reaches the wire.
+            PromptReceipt receipt = client.submitTurn(
+                    SubmitHarnessTurn.builder()
+                            .session(session)
+                            .promptId(SECOND_PROMPT_ID)
+                            .addContent(block)
+                            .payloadDigest(
+                                    SubmitHarnessTurn.computePayloadDigest(
+                                            List.of(block)))
+                            .build());
+            assertEquals(SECOND_PROMPT_ID, receipt.getPromptId());
+        }
+
+        assertEquals(3, promptCalls.get());
+    }
+
    @Test
    void aRefusedCloseByIdKeepsStateRegisteredMidFlight() throws Exception {
        // R5-3: a by-id close may retire only the records it observed
@@ -5463,6 +5574,103 @@ class HostedHarnessClientTest {
        }
    }

+    @Test
+    void aRetryBeginningBetweenTheStatusSnapshotReadsKeepsTheMarker()
+            throws Exception {
+        // R17-3: getStatus reads the marker's clock before its settledness
+        // and re-checks both inside the bin-locked clear, so a
+        // same-identity retry whose beginSend lands between the two
+        // snapshot reads keeps the marker — the status answer the server
+        // evaluated before the retry's send landed must not clear it.
+        createSessionRoute();
+        CountDownLatch sawRetry = new CountDownLatch(1);
+        CountDownLatch releaseRetry = new CountDownLatch(1);
+        AtomicInteger promptCalls = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/prompt",
+                exchange -> {
+                    int call = promptCalls.incrementAndGet();
+                    if (call == 1) {
+                        sendSessionJson(exchange, 503, "{\"error\":\"x\"}");
+                        return;
+                    }
+                    if (call == 2) {
+                        sawRetry.countDown();
+                        try {
+                            releaseRetry.await(15, TimeUnit.SECONDS);
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                    sendSessionJson(exchange, 202,
+                            "{\"promptId\":\"" + (call == 2 ? PROMPT_ID
+                                    : SECOND_PROMPT_ID)
+                                    + "\",\"lastEventId\":0,"
+                                    + "\"eventEpoch\":\""
+                                    + EVENT_EPOCH + "\"}");
+                });
+        server.createContext("/session/" + SESSION_ID + "/status",
+                exchange -> sendSessionJson(exchange, 200,
+                        "{\"sessionId\":\"" + SESSION_ID
+                                + "\",\"hasActivePrompt\":false}"));
+        Map<String, Object> block = Map.of(
+                "type", "text", "text", "status-snapshot-race");
+        String digest = SubmitHarnessTurn.computePayloadDigest(
+                List.of(block));
+
+        HostedHarnessClient client = newClient();
+        ExecutorService callers = Executors.newSingleThreadExecutor();
+        try {
+            HarnessSessionRef session = createSession(client);
+            SubmitHarnessTurn first = SubmitHarnessTurn.builder()
+                    .session(session)
+                    .promptId(PROMPT_ID)
+                    .addContent(block)
+                    .payloadDigest(digest)
+                    .build();
+            // Outcome-unknown: the marker is retained and settled.
+            assertThrows(PromptAdmissionUnknownException.class,
+                    () -> client.submitTurn(first));
+            AtomicReference<Future<?>> retry = new AtomicReference<>();
+            client.statusSnapshotProbe = () -> {
+                // The seam between the two snapshot reads: the retry's
+                // beginSend lands here, after the clock read and before
+                // the settledness read.
+                retry.set(callers.submit(() -> {
+                    client.submitTurn(first);
+                    return null;
+                }));
+                try {
+                    assertTrue(sawRetry.await(5, TimeUnit.SECONDS));
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            };
+            // The retry's send is parked: the status answer must not clear
+            // the marker underneath its open admission window.
+            assertFalse(client.getStatus(session).hasActivePrompt());
+            client.statusSnapshotProbe = () -> {};
+            releaseRetry.countDown();
+            retry.get().get(5, TimeUnit.SECONDS);
+            // The re-admitted marker is live: a different identity is
+            // vetoed locally and never reaches the wire.
+            DaemonException veto = assertThrows(DaemonException.class,
+                    () -> client.submitTurn(SubmitHarnessTurn.builder()
+                            .session(session)
+                            .promptId(SECOND_PROMPT_ID)
+                            .addContent(block)
+                            .payloadDigest(digest)
+                            .build()));
+            assertEquals("Hosted Harness session already has a running turn",
+                    veto.getMessage());
+            assertEquals(2, promptCalls.get());
+        } finally {
+            client.statusSnapshotProbe = () -> {};
+            releaseRetry.countDown();
+            callers.shutdownNow();
+            client.close();
+        }
+    }
+
    @Test
    void aStaleRefCloseKeepsTheCurrentOwnersMarker() {
        // R3-2 (e): the close twin of the stale-detach entrance — the
@@ -5641,14 +5849,18 @@ class HostedHarnessClientTest {
    }

    @Test
-    void aSuccessfulCloseByIdRetiresAnAttachmentRegisteredMidFlight()
+    void aSuccessfulCloseByIdKeepsAnAttachmentRegisteredMidFlight()
            throws Exception {
-        // R10-2: the by-id close's success arm proves the session deleted
-        // outright, so every attachment for the id is stale — including
-        // one a concurrent loadSession registered while the DELETE was in
-        // flight. A survivor would beat against the deleted session
-        // forever: scheduleHeartbeat swallows the 404, so the periodic
-        // task never suppresses itself.
+        // R17-4: the by-id close's success arm proves the session it
+        // OBSERVED deleted, but the 204 answers a DELETE dispatched before
+        // a concurrent loadSession re-registered the id — the peer still
+        // accepts the replacement's heartbeats, so discarding the
+        // replacement would leave the re-opened session without a
+        // heartbeat for the client's lifetime. The success arm retires the
+        // observed attachment only; a registration whose session is in
+        // fact gone self-retires on its heartbeat's definitive
+        // hosted_session_not_found 404 (pinned by
+        // aSessionNotFoundHeartbeatRetiresTheAttachment).
        createSessionRoute();
        AtomicInteger heartbeats = new AtomicInteger();
        server.createContext("/session/" + SESSION_ID + "/heartbeat",
@@ -5702,15 +5914,109 @@ class HostedHarnessClientTest {
            client.loadSession(new LoadHarnessSession(SESSION_ID));
            releaseDelete.countDown();
            assertNull(close.get(5, TimeUnit.SECONDS));
-            // The successful close deleted the session: the mid-flight
-            // replacement attachment retired with the observed one, so no
-            // heartbeat survives.
+            // The mid-flight replacement attachment postdates the
+            // observation the 204 answers, so it survives the close and
+            // keeps beating.
+            Thread.sleep(100);
+            int beating = heartbeats.get();
+            Thread.sleep(60);
+            assertTrue(heartbeats.get() > beating,
+                    "the re-registered attachment must keep beating");
+        } finally {
+            releaseDelete.countDown();
+            caller.shutdownNow();
+            client.close();
+        }
+    }
+
+    @Test
+    void aCloseByRefKeepsAMidFlightAttachmentUntilItsHeartbeat404()
+            throws Exception {
+        // R16-1: the ref close's deletion arm retires the attachment
+        // identity-matched, so an attachment a concurrent load registered
+        // under a different client id survives it; the session the 204
+        // deleted is gone, so the survivor's heartbeat meets a definitive
+        // hosted_session_not_found 404 and self-retires — the periodic
+        // task leaves the scheduler and no heartbeat is issued again,
+        // instead of beating against the dead session forever.
+        createSessionRoute();
+        AtomicInteger heartbeats = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/heartbeat",
+                exchange -> {
+                    heartbeats.incrementAndGet();
+                    if ("client-2".equals(exchange.getRequestHeaders()
+                            .getFirst(HostedHarnessClient.CLIENT_ID_HEADER))) {
+                        exchange.getResponseHeaders().set(
+                                HostedHarnessClient.BOOT_ID_HEADER, BOOT_ID);
+                        sendJson(exchange, 404,
+                                "{\"code\":\"hosted_session_not_found\"}",
+                                false);
+                        return;
+                    }
+                    sendSessionJson(exchange, 200,
+                            "{\"sessionId\":\"" + SESSION_ID
+                                    + "\",\"clientId\":\"" + CLIENT_ID
+                                    + "\",\"lastSeenAt\":123}");
+                });
+        server.createContext("/session/" + SESSION_ID + "/load",
+                exchange -> {
+                    exchange.getResponseHeaders().set(
+                            HostedHarnessClient.BOOT_ID_HEADER, BOOT_ID);
+                    sendJson(exchange, 200,
+                            sessionJson().replace(CLIENT_ID, "client-2"),
+                            false);
+                });
+        CountDownLatch deleteSeen = new CountDownLatch(1);
+        CountDownLatch releaseDelete = new CountDownLatch(1);
+        server.createContext("/session/" + SESSION_ID, exchange -> {
+            deleteSeen.countDown();
+            try {
+                releaseDelete.await(15, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            assertPrivateHeaders(exchange, true);
+            exchange.getResponseHeaders().set(
+                    HostedHarnessClient.BOOT_ID_HEADER, BOOT_ID);
+            exchange.getResponseHeaders().set("Connection", "close");
+            exchange.sendResponseHeaders(204, -1);
+            exchange.close();
+        });
+
+        HostedHarnessClient client = HostedHarnessClient.builder()
+                .baseUri(baseUri)
+                .bearerToken("harness-token")
+                .capabilityDigest(DIGEST)
+                .heartbeatInterval(Duration.ofMillis(10))
+                .build();
+        ExecutorService caller = Executors.newSingleThreadExecutor();
+        try {
+            HarnessSessionRef session = createSession(client);
+            awaitHeartbeat(heartbeats);
+            Future<Throwable> close = caller.submit(() -> {
+                try {
+                    client.closeSession(session);
+                    return null;
+                } catch (Throwable failure) {
+                    return failure;
+                }
+            });
+            assertTrue(deleteSeen.await(5, TimeUnit.SECONDS));
+            // A concurrent caller re-registers the session under a new
+            // client identity while the DELETE is in flight.
+            client.loadSession(new LoadHarnessSession(SESSION_ID));
+            releaseDelete.countDown();
+            assertNull(close.get(5, TimeUnit.SECONDS));
+            // The close retired only its own identity's attachment; the
+            // survivor's heartbeat met the definitive 404 and
+            // self-retired: the scheduler queue drains and the heartbeats
+            // stop.
+            awaitSchedulerDrained(client);
            Thread.sleep(100);
            int last = heartbeats.get();
            Thread.sleep(60);
            assertEquals(last, heartbeats.get(),
-                    "no heartbeat may survive a close that deleted the"
-                            + " session");
+                    "no heartbeat may survive the definitive 404");
        } finally {
            releaseDelete.countDown();
            caller.shutdownNow();
@@ -5718,6 +6024,48 @@ class HostedHarnessClientTest {
        }
    }

+    @Test
+    void aSessionNotFoundHeartbeatRetiresTheAttachment()
+            throws Exception {
+        // R17-4/R16-1: a heartbeat answered with a definitive fenced 404
+        // hosted_session_not_found proves the session record is gone
+        // server-side — the same proof a deleting close or detach rides —
+        // so the attachment retires itself: it leaves the attachments map
+        // and cancels its periodic task instead of beating against the
+        // dead session for the client's lifetime.
+        createSessionRoute();
+        AtomicInteger heartbeats = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/heartbeat",
+                exchange -> {
+                    heartbeats.incrementAndGet();
+                    exchange.getResponseHeaders().set(
+                            HostedHarnessClient.BOOT_ID_HEADER, BOOT_ID);
+                    sendJson(exchange, 404,
+                            "{\"code\":\"hosted_session_not_found\"}",
+                            false);
+                });
+
+        HostedHarnessClient client = HostedHarnessClient.builder()
+                .baseUri(baseUri)
+                .bearerToken("harness-token")
+                .capabilityDigest(DIGEST)
+                .heartbeatInterval(Duration.ofMillis(10))
+                .build();
+        try {
+            createSession(client);
+            // The first heartbeat self-retires the attachment: the
+            // scheduler queue drains and no further heartbeat is issued.
+            awaitSchedulerDrained(client);
+            Thread.sleep(100);
+            int last = heartbeats.get();
+            Thread.sleep(60);
+            assertEquals(last, heartbeats.get(),
+                    "no heartbeat may be issued after the definitive 404");
+        } finally {
+            client.close();
+        }
+    }
+
    @ParameterizedTest
    @MethodSource("closeAndDetachOverloads")
    void aTurnActiveCloseThenLifecycleDetachReleasesTheLedger(boolean byId)
@@ -6847,6 +7195,49 @@ class HostedHarnessClientTest {
        assertEquals(1, promptCalls.get());
    }

+    @Test
+    void anUnfencedCodelessConflictIsOutcomeUnknown() {
+        // R15-3: an unfenced 409 whose body carries no refusal-vocabulary
+        // code is not the route's answer — the intermediary that stripped
+        // the boot header produced it — so it cannot prove this submission
+        // was refused: the outcome is unknown and the owned marker stays
+        // vetoing until a status read. Surfacing it as
+        // DaemonHttpException(409) instead would let the turn consumer
+        // retry the identical refused submission forever.
+        createSessionRoute();
+        AtomicInteger promptCalls = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/prompt",
+                exchange -> {
+                    promptCalls.incrementAndGet();
+                    sendJson(exchange, 409, "{\"error\":\"conflict\"}",
+                            false);
+                });
+        Map<String, Object> block = Map.of(
+                "type", "text", "text", "unfenced-codeless-conflict");
+
+        try (HostedHarnessClient client = newClient()) {
+            HarnessSessionRef session = createSession(client);
+            assertThrows(PromptAdmissionUnknownException.class,
+                    () -> client.submitTurn(requestForSession(block,
+                            session)));
+            // The retained marker still owns the session: a different
+            // identity is vetoed locally and never reaches the wire.
+            DaemonException veto = assertThrows(DaemonException.class,
+                    () -> client.submitTurn(SubmitHarnessTurn.builder()
+                            .session(session)
+                            .promptId(SECOND_PROMPT_ID)
+                            .addContent(block)
+                            .payloadDigest(
+                                    SubmitHarnessTurn.computePayloadDigest(
+                                            List.of(block)))
+                            .build()));
+            assertEquals("Hosted Harness session already has a running turn",
+                    veto.getMessage());
+        }
+
+        assertEquals(1, promptCalls.get());
+    }
+
    @Test
    void overlappingSameIdentityRefusalsReleaseTheMarker()
            throws Exception {
@@ -6958,6 +7349,106 @@ class HostedHarnessClientTest {
        }
    }

+    @Test
+    void aRetryDropThenAnOwnerRefusalKeepsTheMarker() throws Exception {
+        // R15-4: the reverse conclusion order of
+        // overlappingSameIdentityRefusalsReleaseTheMarker — the joined
+        // retry's outcome-unknown drop concludes FIRST (blocking the owed
+        // release), then the owner's definitive refusal concludes the
+        // entry. The owner's settled release must honour the block: this
+        // promptId may be admitted server-side, so the marker keeps
+        // vetoing until a status read clears it.
+        createSessionRoute();
+        CountDownLatch ownerSeen = new CountDownLatch(1);
+        CountDownLatch releaseOwner = new CountDownLatch(1);
+        AtomicInteger promptCalls = new AtomicInteger();
+        server.createContext("/session/" + SESSION_ID + "/prompt",
+                exchange -> {
+                    int call = promptCalls.incrementAndGet();
+                    if (call == 1) {
+                        ownerSeen.countDown();
+                        try {
+                            releaseOwner.await(15, TimeUnit.SECONDS);
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                        sendSessionJson(exchange, 400,
+                                "{\"code\":\"bad_prompt\"}");
+                        return;
+                    }
+                    if (call == 2) {
+                        // The retry's send is dropped before any answer.
+                        exchange.close();
+                        return;
+                    }
+                    sendSessionJson(exchange, 202,
+                            "{\"promptId\":\"" + SECOND_PROMPT_ID
+                                    + "\",\"lastEventId\":0,"
+                                    + "\"eventEpoch\":\""
+                                    + EVENT_EPOCH + "\"}");
+                });
+        Map<String, Object> block = Map.of(
+                "type", "text", "text", "retry-drop-then-refusal");
+        String digest = SubmitHarnessTurn.computePayloadDigest(
+                List.of(block));
+
+        HostedHarnessClient client = newClient();
+        ExecutorService callers = Executors.newFixedThreadPool(2);
+        try {
+            HarnessSessionRef session = createSession(client);
+            SubmitHarnessTurn first = SubmitHarnessTurn.builder()
+                    .session(session)
+                    .promptId(PROMPT_ID)
+                    .addContent(block)
+                    .payloadDigest(digest)
+                    .build();
+            Future<Throwable> owner = callers.submit(() -> {
+                try {
+                    client.submitTurn(first);
+                    return null;
+                } catch (Throwable failure) {
+                    return failure;
+                }
+            });
+            assertTrue(ownerSeen.await(5, TimeUnit.SECONDS));
+            Future<Throwable> retry = callers.submit(() -> {
+                try {
+                    client.submitTurn(first);
+                    return null;
+                } catch (Throwable failure) {
+                    return failure;
+                }
+            });
+            // The retry's dropped send concludes outcome-unknown first.
+            Throwable retryFailure = retry.get(5, TimeUnit.SECONDS);
+            assertTrue(retryFailure
+                            instanceof PromptAdmissionUnknownException,
+                    String.valueOf(retryFailure));
+            // The owner's definitive refusal concludes the entry: the
+            // blocked release keeps the marker.
+            releaseOwner.countDown();
+            Throwable ownerFailure = owner.get(5, TimeUnit.SECONDS);
+            assertTrue(ownerFailure instanceof DaemonHttpException,
+                    String.valueOf(ownerFailure));
+            assertEquals(400,
+                    ((DaemonHttpException) ownerFailure).getStatusCode());
+            DaemonException veto = assertThrows(DaemonException.class,
+                    () -> client.submitTurn(SubmitHarnessTurn.builder()
+                            .session(session)
+                            .promptId(SECOND_PROMPT_ID)
+                            .addContent(block)
+                            .payloadDigest(digest)
+                            .build()));
+            assertEquals("Hosted Harness session already has a running turn",
+                    veto.getMessage());
+            assertEquals(2, promptCalls.get());
+        } finally {
+            releaseOwner.countDown();
+            callers.shutdownNow();
+            client.close();
+        }
+    }
+
    @Test
    void anOwnerReleaseDuringARetryRepublishKeepsTheLedger()
            throws Exception {