commit 2763d2f18e89ea418bb22be61d26cc0042d9a211 Author: Codex AuthorDate: Sat Oct 31 22:42:43 2026 +0800 Commit: Codex CommitDate: Sat Oct 31 22:42:43 2026 +0800 Retain late semantic replies for validated reuse on later reviews diff --git a/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java b/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java index f673249..0080e3f 100644 --- a/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java +++ b/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java @@ -21,6 +21,7 @@ import com.android.commands.monkey.ape.model.ModelAction; import com.android.commands.monkey.ape.model.State; import com.android.commands.monkey.ape.tree.GUITree; import com.android.commands.monkey.ape.tree.GUITreeNode; +import com.android.commands.monkey.ape.utils.Config; /** Page-local semantic projection. Registry remains the only execution history. */ public final class JevSemanticReview implements AutoCloseable { @@ -52,6 +53,7 @@ public final class JevSemanticReview implements AutoCloseable { final String sourceGui; final Map decisions=new HashMap(); boolean valid; + long sourceVersion,historyRevision; Cache(Map c,JevSemanticProtocol.Request r,JevTransport.Call q,int a,String gui){candidates=c;request=r;call=q;attempt=a;sourceGui=gui; byId=new HashMap();for(Candidate candidate:c.values())byId.put(candidate.candidateId,candidate);} } @@ -60,9 +62,12 @@ public final class JevSemanticReview implements AutoCloseable { private final JevSemanticNavigationMask navigationMask; private final String model; private final String endpoint; - private final int timeoutMs,maxBytes; + private final int timeoutMs,maxBytes,requestLifetimeMs,retainedCapacity; private final AtomicLong sequence=new AtomicLong(); private final Map pages=new HashMap(); + // Completed raw responses stay on their Call. Only the APE thread may validate/apply them. + private final Map retained=new LinkedHashMap(); + private long modelVersion=-1; private final Set inFlight=new HashSet(); private final Set served=new HashSet(); private final String contract; @@ -87,15 +92,17 @@ public final class JevSemanticReview implements AutoCloseable { public JevSemanticReview(String endpoint,String token,int queueCapacity,int timeoutMs,int maxBytes,String model,JevRegistry history,ReceiptSink sink,boolean waitForResponse)throws IOException { transport=new JevTransport(endpoint,token,queueCapacity,"probe"); registry=history;navigationMask=new JevSemanticNavigationMask(history);this.endpoint=endpoint;this.model=model;this.contract="semantic-v1:"+model;this.timeoutMs=timeoutMs;this.maxBytes=maxBytes;receiptSink=sink;synchronous=waitForResponse; + requestLifetimeMs=Math.max(timeoutMs,Config.getInteger("ape.jev.timeoutMs",60000)); + retainedCapacity=queueCapacity+1; } public boolean isSynchronous(){return synchronous;} public boolean isHealthy(){return healthy;} public boolean hasLocalFailure(){return localFailure;} public void setStep(long step){currentStep=step;} public void setModelVersion(long version) {reviewLock.lock();try { - // APE refinement changes graph objects, not the decision contract. Keep - // completed and in-flight reviews addressable by concrete identity. - // The value is intentionally not part of a page or ledger key. + modelVersion=version; + // Existing terminal ledger semantics stay unchanged. Unapplied retained + // responses require this exact version again before they can be consumed. } finally {reviewLock.unlock();}} public void setSnapshotIdentity(Map identity) {reviewLock.lock();try { transport.setSnapshotIdentity(identity); } finally {reviewLock.unlock();}} public void markExecuted(State state,ModelAction action,GUITreeNode node) {reviewLock.lock();try { @@ -121,7 +128,7 @@ public final class JevSemanticReview implements AutoCloseable { } catch(InterruptedException e) {Thread.currentThread().interrupt();healthy=false;selectionBlocked=true;return;} try {currentStep=step;prepareLocked(state,tree,absolute);} catch(JevDeadline.Expired e) { - for(Cache c:pages.values())c.call.cancel("DEADLINE"); + for(Cache cache:pages.values())retain(cache); pages.clear();inFlight.clear(); healthy=false;selectionBlocked=true; try {failOpen(state,"DEADLINE");}catch(JevDeadline.Expired ignored) { /* audit lock busy: do not wait again */ } @@ -135,6 +142,8 @@ public final class JevSemanticReview implements AutoCloseable { if(!synchronous)poll(); Page page=page(state,tree); if(page==null||page.candidates.isEmpty())return; + int retries=synchronous?resumeRetained(state,tree,page,deadline):0; + if(retries<0)return; Map pending=new LinkedHashMap(); for(Map.Entry entry:page.candidates.entrySet()) if(registry.reviewLedger().get(contract,scope(entry.getValue()),identity(entry.getValue()))==null&&!inFlight.contains(entry.getKey()))pending.put(entry.getKey(),entry.getValue()); @@ -154,7 +163,6 @@ public final class JevSemanticReview implements AutoCloseable { requestPage=new Page(page.cacheKey+"|pending",pendingPayload(page.payload,fingerprintMisses),fingerprintMisses); request=JevSemanticProtocol.request("semantic-"+sequence.incrementAndGet(),model,requestPage.payload,maxBytes); } - int retries=0; while(true) { long remaining=deadline-System.nanoTime(); if(synchronous && (remaining<=0 || Thread.currentThread().isInterrupted())) { @@ -162,12 +170,17 @@ public final class JevSemanticReview implements AutoCloseable { } JevDeadline.check(); int attempt=nextAttempt(request.inputFingerprint); + // Capture before submitting, so expiry cannot orphan an accepted call. + String sourceGui=JevEvidence.guiIdentity(tree); + long historyRevision=registry.executionRevision(state); receipt("semantic-request-"+request.requestId,requestReceipt(request,requestPage.candidates,attempt)); registry.semanticEvent("JEV_REQUEST_SENT",currentStep,state.getGraphId(),null,JevRegistry.fields( "request_id",request.requestId,"candidate_count",requestPage.candidates.size(),"wait_mode",synchronous?"synchronous":"asynchronous", "attempt",attempt,"input_fingerprint",request.inputFingerprint)); JevDeadline.check(); - JevTransport.Call call=transport.submit(request.requestId,request.bodyJson,synchronous?(int)Math.max(1,java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(Math.max(0,deadline-System.nanoTime()))):timeoutMs,Math.max(16384,maxBytes)); + JevTransport.Call call=synchronous + ?transport.submitRetained(request.requestId,request.bodyJson,requestLifetimeMs,Math.max(16384,maxBytes)) + :transport.submit(request.requestId,request.bodyJson,timeoutMs,Math.max(16384,maxBytes)); if(call==null) { requestFailure(request,attempt,"QUEUE_FULL",0,null); if(retries++=MAX_RETRIES){failOpen(state,"RETRIES_EXHAUSTED");return;} @@ -200,6 +215,74 @@ public final class JevSemanticReview implements AutoCloseable { registry.semanticEvent("SEMANTIC_FAIL_OPEN",currentStep,state.getGraphId(),null, JevRegistry.fields("reason",reason,"fallback","native","semantic_selection_blocked",true)); } + private void retain(Cache cache) { + retained.put(cache.request.requestId,cache); + while(retained.size()>retainedCapacity) { + Iterator> it=retained.entrySet().iterator(); + Cache evicted=it.next().getValue();it.remove();evicted.call.cancel("RETAINED_CAPACITY"); + } + } + /** Archive raw bytes even if APE never revisits. No decisions are parsed or applied here. */ + private void archiveOnArrival(Cache cache,final long sourceStep) { + if(receiptSink==null)return; + final JevSemanticProtocol.Request request=cache.request; + final long sourceVersion=cache.sourceVersion,historyRevision=cache.historyRevision; + final String sourceGui=cache.sourceGui; + cache.call.onComplete(new JevTransport.Completion(){public void completed(final JevTransport.Result result) { + try {receiptWorker.execute(new Runnable(){public void run() { + try { + JSONObject value=new JSONObject().put("schema",1).put("kind","semantic_provider_arrival") + .put("request_id",request.requestId).put("endpoint",endpoint).put("model",model) + .put("contract",contract).put("source_step",sourceStep).put("source_version",sourceVersion) + .put("history_revision",historyRevision).put("source_gui",sourceGui) + .put("input_signature",request.inputSignature).put("input_fingerprint",request.inputFingerprint) + .put("http_status",result.httpStatus).put("error",result.error==null?JSONObject.NULL:result.error) + .put("raw_response",result.raw==null?JSONObject.NULL:result.raw) + .put("submitted_ns",result.submittedNanos).put("sent_ns",result.sentNanos) + .put("completed_ns",result.completedNanos).put("wait_ns",result.waitNanos()) + .put("validated",false).put("applied",false); + if(!receiptSink.save("semantic-arrival-"+request.requestId,value))healthy=false; + }catch(JSONException e){healthy=false;}catch(RuntimeException e){healthy=false;} + }});}catch(RejectedExecutionException e){healthy=false;} + }}); + } + /** A later review may reuse raw work only against the same visible input and history. */ + private int resumeRetained(State state,GUITree tree,Page page,long deadline) { + for(Map.Entry item:new ArrayList>(retained.entrySet())) { + JevDeadline.check();Cache old=item.getValue(); + if(old.sourceVersion!=modelVersion) { + old.call.cancel("STALE_HISTORY_OR_VERSION");retained.remove(item.getKey());continue; + } + JevTransport.Result response=old.call.poll(); + if(!old.request.inputSignature.equals(page.payload.optString("input_signature"))) { + if(response!=null && response.error!=null)retained.remove(item.getKey()); + continue; + } + if(old.historyRevision!=registry.executionRevision(state)) { + old.call.cancel("STALE_HISTORY_OR_VERSION");retained.remove(item.getKey());continue; + } + if(!old.sourceGui.equals(JevEvidence.guiIdentity(tree)))continue; + Map current=new LinkedHashMap(); + for(String key:old.candidates.keySet()) { + Candidate c=page.candidates.get(key); + if(c!=null && old.byId.containsKey(c.candidateId) && old.byId.get(c.candidateId).key.equals(key))current.put(key,c); + } + if(current.size()!=old.request.candidateIds.size()) { + old.call.cancel("STALE_CANDIDATES");retained.remove(item.getKey());continue; + } + Cache rebound=new Cache(current,old.request,old.call,old.attempt,JevEvidence.guiIdentity(tree)); + rebound.sourceVersion=old.sourceVersion;rebound.historyRevision=old.historyRevision; + // Keep it recoverable even if parsing/receipt processing exhausts this later budget. + pages.put(item.getKey(),rebound);retained.remove(item.getKey()); + if(response==null)response=old.call.awaitUntil(deadline); + JevDeadline.check(); + if(response==null)throw new JevDeadline.Expired(); + if(consumeCache(item.getKey(),rebound,response))return -1; + // Protocol / HTTP failures remain failures, rather than reusable semantic answers. + return 1; + } + return 0; + } private static final class Page { final String cacheKey;final JSONObject payload;final Map candidates; Page(String k,JSONObject p,Map c){cacheKey=k;payload=p;candidates=c;} @@ -442,6 +525,7 @@ public final class JevSemanticReview implements AutoCloseable { return new JSONObject().put("schema",1).put("kind",synchronous?"sync_semantic_request":"async_semantic_request").put("request_id",request.requestId) .put("attempt",attempt).put("endpoint",endpoint).put("model",model).put("contract",contract).put("step",currentStep) .put("retry_count",attempt-1) + .put("request_lifetime_ms",synchronous?requestLifetimeMs:timeoutMs).put("late_response_reuse",synchronous) .put("input_signature",request.inputSignature).put("input_fingerprint",request.inputFingerprint).put("candidate_ids",ids) .put("request_bytes",request.requestBytes).put("timeout_ms",timeoutMs).put("body",new JSONObject(request.bodyJson)); } catch(JSONException error) { return new JSONObject(); } @@ -475,5 +559,5 @@ public final class JevSemanticReview implements AutoCloseable { "scrollable",node.isScrollable(),"password",node.isPassword(),"input_text",JevEvidence.boundedText(node.getInputText()), "bounds",new JSONArray().put(b.left).put(b.top).put(b.right).put(b.bottom),"role_signature",JevEvidence.roleSignature(node)); } - public void close(){reviewLock.lock();try {receiptWorker.shutdownNow();transport.close();pages.clear();inFlight.clear();served.clear();} finally {reviewLock.unlock();}} + public void close(){reviewLock.lock();try {receiptWorker.shutdownNow();transport.close();pages.clear();retained.clear();inFlight.clear();served.clear();} finally {reviewLock.unlock();}} } diff --git a/native/src/com/android/commands/monkey/ape/jev/JevTransport.java b/native/src/com/android/commands/monkey/ape/jev/JevTransport.java index f35486c..beed490 100644 --- a/native/src/com/android/commands/monkey/ape/jev/JevTransport.java +++ b/native/src/com/android/commands/monkey/ape/jev/JevTransport.java @@ -13,6 +13,7 @@ import java.util.LinkedHashMap; /** One bounded HTTP worker. Deadlines include the queue and never wait for socket cleanup. */ public final class JevTransport implements Closeable { + interface Completion { void completed(Result result); } public static final class Result { public final String raw,error; public final int httpStatus; @@ -33,16 +34,26 @@ public final class JevTransport implements Closeable { private volatile HttpURLConnection connection; private final java.util.concurrent.atomic.AtomicReference result=new java.util.concurrent.atomic.AtomicReference(); private final java.util.concurrent.atomic.AtomicBoolean consumed=new java.util.concurrent.atomic.AtomicBoolean(); + private final java.util.concurrent.atomic.AtomicBoolean notified=new java.util.concurrent.atomic.AtomicBoolean(); + private volatile Completion completion; private final java.util.concurrent.CountDownLatch ready=new java.util.concurrent.CountDownLatch(1); private TimerTask expiry; - private Call(String id,String body,int timeout,int limit) { + private Call(String id,String body,int timeout,int limit,boolean retainAfterWait) { requestId=id;this.body=body.getBytes(StandardCharsets.UTF_8);responseLimit=limit; - deadline=Math.min(JevDeadline.absolute(),submitted+TimeUnit.MILLISECONDS.toNanos(timeout)); + long requestDeadline=submitted+TimeUnit.MILLISECONDS.toNanos(timeout); + deadline=retainAfterWait?requestDeadline:Math.min(JevDeadline.absolute(),requestDeadline); } private boolean finish(String raw,String error,int status) { if(!result.compareAndSet(null,new Result(this,raw,error,status)))return false; ready.countDown(); - if(expiry!=null)expiry.cancel();return true; + if(expiry!=null)expiry.cancel();notifyCompletion();return true; + } + // The listener only enqueues immutable receipt data; it must never wait or touch GUI state. + void onComplete(Completion listener) {completion=listener;notifyCompletion();} + private void notifyCompletion() { + Completion listener=completion;Result value=result.get(); + if(listener!=null && value!=null && notified.compareAndSet(false,true)) + listener.completed(value); } public Result poll() { if(result.get()==null && System.nanoTime()>=deadline)cancel("TIMEOUT"); @@ -59,6 +70,13 @@ public final class JevTransport implements Closeable { } catch(InterruptedException interrupted){cancel("INTERRUPTED");Thread.currentThread().interrupt();} return result.get(); } + /** End the caller's wait without cancelling an already submitted request. */ + public Result awaitUntil(long waitDeadline) { + if(poll()==null)try { + ready.await(Math.max(0,Math.min(waitDeadline,deadline)-System.nanoTime()),TimeUnit.NANOSECONDS); + }catch(InterruptedException e){Thread.currentThread().interrupt();} + return poll(); + } public void cancel(String reason) { if(finish(null,reason==null?"CANCELLED":reason,0)) { waiting.remove(this); @@ -153,12 +171,19 @@ public final class JevTransport implements Closeable { catch(RuntimeException ignored){ } }},"ape-"+requestClass+"-cleanup");cleanup.setDaemon(true);cleanup.start(); } - public Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes) {JevDeadline.lock(submitLock);try { + public Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes) { + return submit(requestId,body,timeoutMs,maxResponseBytes,false); + } + /** Only the network lifetime outlives the caller's budget; enqueue remains budgeted. */ + Call submitRetained(String requestId,String body,int timeoutMs,int maxResponseBytes) { + return submit(requestId,body,timeoutMs,maxResponseBytes,true); + } + private Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes,boolean retained) {JevDeadline.lock(submitLock);try { if(requestId==null || requestId.isEmpty() || body==null || timeoutMs<=0 || maxResponseBytes<=0) throw new IllegalArgumentException("invalid request bounds"); if(closed)return null; JevDeadline.check();used=true; - final Call call=new Call(requestId,body,timeoutMs,maxResponseBytes); + final Call call=new Call(requestId,body,timeoutMs,maxResponseBytes,retained); call.expiry=new TimerTask(){public void run(){call.cancel("TIMEOUT");}}; // Schedule before publishing: an immediate response may cancel this task. deadlines.schedule(call.expiry,Math.max(1,TimeUnit.NANOSECONDS.toMillis(Math.max(0,call.deadline-System.nanoTime()))));