Dai Ha
541df87272
A backend that refuses on a subscription usage limit leaves the pane healthy but the turn ends with no bridge_reply; the completion fallback used to scrape and hand that refusal back as if it were a real answer. CompletionResolver now matches the scrape against a per-profile exhaustedPattern (config, never a vendor string) and resolves the send as Rendezvous.Kind/Outcome.BACKEND_EXHAUSTED with a reason carrying the matched line, kept distinct from GONE/WORKER_FAILED. A profile with no pattern configured is unaffected. Coverage is logged at startup via CompletionResolver.coverage(...), naming which profiles have a pattern and which don't, following FleetHealthMonitor.coverage's pattern.
252 lines
13 KiB
Java
252 lines
13 KiB
Java
package dev.ltms.bridged.msg;
|
|
|
|
import java.util.concurrent.CompletableFuture;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
|
|
/**
|
|
* The reply rendezvous: where a blocking {@code bridge_send} awaits how the worker's delegated turn
|
|
* ends. The sending (primary) request thread {@link #open}s a waiter; it is resolved either by the
|
|
* worker's explicit {@code bridge_reply} ({@link #resolve}, arriving on a different thread via
|
|
* {@code POST /sessions/{id}/reply}) or — the CB-106 fallback — by the injector observing the
|
|
* worker's delegated turn return to idle without a reply ({@link #resolveCompletion}).
|
|
*
|
|
* <p>At most one waiter per session — {@link MessageService} serializes sends per session, so a
|
|
* resolution maps unambiguously to the one outstanding send and cannot be captured by another.
|
|
*
|
|
* <p><strong>Waiter identity (CB-116).</strong> The completion/failure fallbacks run asynchronously
|
|
* and can fire <em>after</em> the turn they belong to has already been resolved by an explicit reply
|
|
* and a <em>next</em> send has opened its own waiter on the same session. Resolving "whatever waiter
|
|
* is registered now" would then land turn N's stale scrape on turn N+1's send. So those fallbacks
|
|
* resolve a <em>specific</em> {@link CompletableFuture} captured when their turn was delivered
|
|
* ({@link #resolveCompletion(CompletableFuture, String)} /
|
|
* {@link #resolveFailure(CompletableFuture, String)}): a no-op if that waiter was already resolved,
|
|
* and it can never touch a later send's waiter.
|
|
*/
|
|
public final class Rendezvous {
|
|
|
|
/** How a delegated turn ended (or paused). */
|
|
public enum Kind {
|
|
/** The worker called {@code bridge_reply} with a structured answer. */
|
|
REPLY,
|
|
/** The worker's turn finished without a {@code bridge_reply}; {@code text} is a scrape. */
|
|
COMPLETION,
|
|
/** The worker ran the turn then wedged (CB-109); {@code text} is the failure context. */
|
|
FAILED,
|
|
/**
|
|
* The turn finished without a {@code bridge_reply}, and the scrape matched the backend's
|
|
* configured usage-limit refusal pattern (CB-578 stage A); {@code text} is the reason,
|
|
* carrying the matched line. The pane is healthy — only the account is refusing — so this
|
|
* is kept separate from a session simply going {@code GONE}.
|
|
*/
|
|
BACKEND_EXHAUSTED,
|
|
/**
|
|
* The worker paused mid-turn to ask the primary a question (CB-205 reverse rendezvous);
|
|
* {@code text} is the question and {@code turnId} correlates the primary's answer back to
|
|
* the worker's blocked {@code bridge_ask}. Not terminal — the turn resumes after the answer.
|
|
*/
|
|
QUESTION
|
|
}
|
|
|
|
/**
|
|
* The resolved outcome of a send: its {@link Kind}, the associated text, and — only for
|
|
* {@link Kind#QUESTION} — the {@code turnId} the primary answers with (else {@code null}).
|
|
*/
|
|
public record Resolution(Kind kind, String text, String turnId) {
|
|
/** A terminal resolution (reply / completion / failure) with no correlation id. */
|
|
public Resolution(Kind kind, String text) {
|
|
this(kind, text, null);
|
|
}
|
|
}
|
|
|
|
/** A worker's open mid-turn question: the worker session it belongs to and the answer future. */
|
|
private record AskWaiter(String session, CompletableFuture<String> answer) {
|
|
}
|
|
|
|
/**
|
|
* Handle to a reverse-rendezvous turn: the {@code turnId}, its answer future, and whether this
|
|
* call freshly opened it (versus coalescing onto an already-open ask).
|
|
*/
|
|
public record AskTicket(String turnId, CompletableFuture<String> answer, boolean fresh) {
|
|
}
|
|
|
|
private final ConcurrentHashMap<String, CompletableFuture<Resolution>> waiters = new ConcurrentHashMap<>();
|
|
|
|
/** Reverse rendezvous (CB-205): worker questions awaiting the primary's answer, keyed by {@code turnId}. */
|
|
private final ConcurrentHashMap<String, AskWaiter> asks = new ConcurrentHashMap<>();
|
|
private final AtomicLong askSeq = new AtomicLong();
|
|
/** Per-session index of the currently-open ask, so duplicate bridge_ask calls coalesce onto one turn. */
|
|
private final ConcurrentHashMap<String, String> openAsksBySession = new ConcurrentHashMap<>();
|
|
|
|
/**
|
|
* Register a waiter for {@code session} — the await side of the public {@code resolve*} methods.
|
|
* The caller must hold that session's send lock.
|
|
*
|
|
* <p>Atomic fail-if-present (CB-548): if a waiter is already registered for {@code session}, an
|
|
* {@link IllegalStateException} is thrown rather than replacing the first — so any future
|
|
* invariant violation fails loudly instead of silently swapping the waiter another send is
|
|
* blocked on. {@code MessageService} serializes sends per session (the send lock), so in correct
|
|
* code a double open is impossible; this is a tripwire for the day that no longer holds.
|
|
*/
|
|
public CompletableFuture<Resolution> open(String session) {
|
|
CompletableFuture<Resolution> waiter = new CompletableFuture<>();
|
|
CompletableFuture<Resolution> existing = waiters.putIfAbsent(session, waiter);
|
|
if (existing != null) {
|
|
throw new IllegalStateException(
|
|
"rendezvous double-open for session " + session + " — a waiter is already registered");
|
|
}
|
|
return waiter;
|
|
}
|
|
|
|
/**
|
|
* Remove {@code waiter} for {@code session}, only if it is still the registered one. The
|
|
* symmetric complement of {@link #open}: a terminal send deregisters its waiter so the next
|
|
* send on the session may {@link #open} a fresh one (CB-548 makes double-open an error, so a
|
|
* successful {@code open} after a finished turn requires this close to have happened first).
|
|
*/
|
|
public void close(String session, CompletableFuture<Resolution> waiter) {
|
|
waiters.remove(session, waiter);
|
|
}
|
|
|
|
/** Whether a send is currently awaiting a resolution for {@code session}. */
|
|
public boolean isWaiting(String session) {
|
|
return waiters.containsKey(session);
|
|
}
|
|
|
|
/**
|
|
* The waiter currently registered for {@code session}, or {@code null} if none is waiting. The
|
|
* completion/failure fallbacks capture this at delivery time so they can later resolve that exact
|
|
* send (see the CB-116 note above) rather than whichever send happens to be waiting when they fire.
|
|
*/
|
|
public CompletableFuture<Resolution> currentWaiter(String session) {
|
|
return waiters.get(session);
|
|
}
|
|
|
|
/**
|
|
* Resolve the send awaiting on {@code session} with the worker's explicit reply {@code content}.
|
|
*
|
|
* @return {@code true} if a waiter was resolved; {@code false} if none was waiting (a late or
|
|
* spurious reply — e.g. the send already timed out)
|
|
*/
|
|
public boolean resolve(String session, String content) {
|
|
return complete(session, new Resolution(Kind.REPLY, content));
|
|
}
|
|
|
|
// --- reverse rendezvous (CB-205 bridge_ask) ------------------------------------------------
|
|
|
|
/**
|
|
* Open a reverse-rendezvous waiter for a worker's mid-turn question. If {@code session} already has
|
|
* an open ask, coalesce onto it (same {@code turnId}, same answer future). Otherwise atomically mint
|
|
* a fresh {@code turnId}, register it in both the per-turn and per-session indexes, and hand it back
|
|
* marked fresh. The caller then {@link #resolveQuestion surfaces the question} to the primary and
|
|
* blocks on the returned future until the primary {@link #answerAsk answers}.
|
|
*/
|
|
public AskTicket openAsk(String session) {
|
|
while (true) {
|
|
AskWaiter[] minted = { null };
|
|
String turnId = openAsksBySession.computeIfAbsent(session, _ -> {
|
|
String newTurnId = session + "#" + askSeq.incrementAndGet();
|
|
CompletableFuture<String> answer = new CompletableFuture<>();
|
|
AskWaiter waiter = new AskWaiter(session, answer);
|
|
asks.put(newTurnId, waiter);
|
|
minted[0] = waiter;
|
|
return newTurnId;
|
|
});
|
|
if (minted[0] != null) {
|
|
return new AskTicket(turnId, minted[0].answer(), true);
|
|
}
|
|
AskWaiter existing = asks.get(turnId);
|
|
if (existing != null) {
|
|
return new AskTicket(turnId, existing.answer(), false);
|
|
}
|
|
// A close raced and removed the waiter after we read the turnId; clear the stale index entry
|
|
// and retry so a fresh ask is always backed by a registered waiter.
|
|
openAsksBySession.remove(session, turnId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Surface a worker's mid-turn {@code question} by resolving the primary's open {@code bridge_send}
|
|
* with a {@link Kind#QUESTION} carrying {@code turnId}. Same session-keyed semantics as
|
|
* {@link #resolve}: the one outstanding send for {@code session} unblocks with the question.
|
|
*
|
|
* @return {@code true} if a send was awaiting (the question reached the primary); {@code false}
|
|
* if none was (no delegation is open to answer it)
|
|
*/
|
|
public boolean resolveQuestion(String session, String question, String turnId) {
|
|
return complete(session, new Resolution(Kind.QUESTION, question, turnId));
|
|
}
|
|
|
|
/** The worker session an outstanding ask {@code turnId} belongs to, or {@code null} if unknown/lapsed. */
|
|
public String askSession(String turnId) {
|
|
AskWaiter w = asks.get(turnId);
|
|
return w == null ? null : w.session();
|
|
}
|
|
|
|
/**
|
|
* Resolve a worker's blocked {@code bridge_ask} with the primary's {@code answer}, unblocking it
|
|
* to resume its turn.
|
|
*
|
|
* @return {@code true} if the ask was still open and got the answer; {@code false} if the
|
|
* {@code turnId} is unknown or the ask already lapsed (timed out / was answered)
|
|
*/
|
|
public boolean answerAsk(String turnId, String answer) {
|
|
AskWaiter w = asks.get(turnId);
|
|
return w != null && w.answer().complete(answer);
|
|
}
|
|
|
|
/** Drop a reverse-rendezvous turn once its {@code bridge_ask} has resolved (answered or lapsed). */
|
|
public void closeAsk(String turnId) {
|
|
AskWaiter w = asks.get(turnId);
|
|
if (w == null) {
|
|
return;
|
|
}
|
|
// Remove the session index first and only if it still points to this turn, so a concurrent
|
|
// fresh ask cannot inherit a waiter we are about to drop.
|
|
openAsksBySession.remove(w.session(), turnId);
|
|
asks.remove(turnId);
|
|
}
|
|
|
|
/**
|
|
* Resolve a specific captured {@code waiter} as a completion (the delegated turn finished with no
|
|
* {@code bridge_reply}); {@code text} is the scraped transcript tail. The waiter is the one
|
|
* captured when this turn was delivered, so a late completion for turn N cannot land on turn N+1's
|
|
* send (CB-116). A no-op if that waiter was already resolved — a raced {@code bridge_reply} wins.
|
|
*
|
|
* @return {@code true} if this call resolved the waiter, {@code false} if it was null or already resolved
|
|
*/
|
|
public boolean resolveCompletion(CompletableFuture<Resolution> waiter, String text) {
|
|
return waiter != null && waiter.complete(new Resolution(Kind.COMPLETION, text));
|
|
}
|
|
|
|
/**
|
|
* Resolve a specific captured {@code waiter} as a failure — the worker ran the turn but wedged in
|
|
* an unrecoverable state (CB-109); {@code reason} is the failure context (e.g. the error screen).
|
|
* Like {@link #resolveCompletion(CompletableFuture, String)} it targets the exact captured send
|
|
* (CB-116). A no-op if that waiter was already resolved — first resolution wins.
|
|
*
|
|
* @return {@code true} if this call resolved the waiter, {@code false} if it was null or already resolved
|
|
*/
|
|
public boolean resolveFailure(CompletableFuture<Resolution> waiter, String reason) {
|
|
return waiter != null && waiter.complete(new Resolution(Kind.FAILED, reason));
|
|
}
|
|
|
|
/**
|
|
* Resolve a specific captured {@code waiter} as {@link Kind#BACKEND_EXHAUSTED} (CB-578 stage A):
|
|
* the turn finished with no {@code bridge_reply} and the scrape matched the backend's configured
|
|
* usage-limit pattern; {@code reason} carries the matched line. Like
|
|
* {@link #resolveCompletion(CompletableFuture, String)} it targets the exact captured send
|
|
* (CB-116). A no-op if that waiter was already resolved — first resolution wins.
|
|
*
|
|
* @return {@code true} if this call resolved the waiter, {@code false} if it was null or already resolved
|
|
*/
|
|
public boolean resolveExhausted(CompletableFuture<Resolution> waiter, String reason) {
|
|
return waiter != null && waiter.complete(new Resolution(Kind.BACKEND_EXHAUSTED, reason));
|
|
}
|
|
|
|
private boolean complete(String session, Resolution resolution) {
|
|
CompletableFuture<Resolution> waiter = waiters.get(session);
|
|
return waiter != null && waiter.complete(resolution);
|
|
}
|
|
}
|