design-2763d2f.diff
原始文件
1commit 2763d2f18e89ea418bb22be61d26cc0042d9a211
2Author: Codex <codex@local>
3AuthorDate: Sat Oct 31 22:42:43 2026 +0800
4Commit: Codex <codex@local>
5CommitDate: Sat Oct 31 22:42:43 2026 +0800
6
7 Retain late semantic replies for validated reuse on later reviews
8
9diff --git a/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java b/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java
10index f673249..0080e3f 100644
11--- a/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java
12+++ b/native/src/com/android/commands/monkey/ape/jev/JevSemanticReview.java
13@@ -21,6 +21,7 @@ import com.android.commands.monkey.ape.model.ModelAction;
14 import com.android.commands.monkey.ape.model.State;
15 import com.android.commands.monkey.ape.tree.GUITree;
16 import com.android.commands.monkey.ape.tree.GUITreeNode;
17+import com.android.commands.monkey.ape.utils.Config;
18
19 /** Page-local semantic projection. Registry remains the only execution history. */
20 public final class JevSemanticReview implements AutoCloseable {
21@@ -52,6 +53,7 @@ public final class JevSemanticReview implements AutoCloseable {
22 final String sourceGui;
23 final Map<String,JevSemanticProtocol.Decision> decisions=new HashMap<String,JevSemanticProtocol.Decision>();
24 boolean valid;
25+ long sourceVersion,historyRevision;
26 Cache(Map<String,Candidate> c,JevSemanticProtocol.Request r,JevTransport.Call q,int a,String gui){candidates=c;request=r;call=q;attempt=a;sourceGui=gui;
27 byId=new HashMap<String,Candidate>();for(Candidate candidate:c.values())byId.put(candidate.candidateId,candidate);}
28 }
29@@ -60,9 +62,12 @@ public final class JevSemanticReview implements AutoCloseable {
30 private final JevSemanticNavigationMask navigationMask;
31 private final String model;
32 private final String endpoint;
33- private final int timeoutMs,maxBytes;
34+ private final int timeoutMs,maxBytes,requestLifetimeMs,retainedCapacity;
35 private final AtomicLong sequence=new AtomicLong();
36 private final Map<String,Cache> pages=new HashMap<String,Cache>();
37+ // Completed raw responses stay on their Call. Only the APE thread may validate/apply them.
38+ private final Map<String,Cache> retained=new LinkedHashMap<String,Cache>();
39+ private long modelVersion=-1;
40 private final Set<String> inFlight=new HashSet<String>();
41 private final Set<String> served=new HashSet<String>();
42 private final String contract;
43@@ -87,15 +92,17 @@ public final class JevSemanticReview implements AutoCloseable {
44 public JevSemanticReview(String endpoint,String token,int queueCapacity,int timeoutMs,int maxBytes,String model,JevRegistry history,ReceiptSink sink,boolean waitForResponse)throws IOException {
45 transport=new JevTransport(endpoint,token,queueCapacity,"probe");
46 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;
47+ requestLifetimeMs=Math.max(timeoutMs,Config.getInteger("ape.jev.timeoutMs",60000));
48+ retainedCapacity=queueCapacity+1;
49 }
50 public boolean isSynchronous(){return synchronous;}
51 public boolean isHealthy(){return healthy;}
52 public boolean hasLocalFailure(){return localFailure;}
53 public void setStep(long step){currentStep=step;}
54 public void setModelVersion(long version) {reviewLock.lock();try {
55- // APE refinement changes graph objects, not the decision contract. Keep
56- // completed and in-flight reviews addressable by concrete identity.
57- // The value is intentionally not part of a page or ledger key.
58+ modelVersion=version;
59+ // Existing terminal ledger semantics stay unchanged. Unapplied retained
60+ // responses require this exact version again before they can be consumed.
61 } finally {reviewLock.unlock();}}
62 public void setSnapshotIdentity(Map<String,String> identity) {reviewLock.lock();try { transport.setSnapshotIdentity(identity); } finally {reviewLock.unlock();}}
63 public void markExecuted(State state,ModelAction action,GUITreeNode node) {reviewLock.lock();try {
64@@ -121,7 +128,7 @@ public final class JevSemanticReview implements AutoCloseable {
65 } catch(InterruptedException e) {Thread.currentThread().interrupt();healthy=false;selectionBlocked=true;return;}
66 try {currentStep=step;prepareLocked(state,tree,absolute);}
67 catch(JevDeadline.Expired e) {
68- for(Cache c:pages.values())c.call.cancel("DEADLINE");
69+ for(Cache cache:pages.values())retain(cache);
70 pages.clear();inFlight.clear();
71 healthy=false;selectionBlocked=true;
72 try {failOpen(state,"DEADLINE");}catch(JevDeadline.Expired ignored) { /* audit lock busy: do not wait again */ }
73@@ -135,6 +142,8 @@ public final class JevSemanticReview implements AutoCloseable {
74 if(!synchronous)poll();
75 Page page=page(state,tree);
76 if(page==null||page.candidates.isEmpty())return;
77+ int retries=synchronous?resumeRetained(state,tree,page,deadline):0;
78+ if(retries<0)return;
79 Map<String,Candidate> pending=new LinkedHashMap<String,Candidate>();
80 for(Map.Entry<String,Candidate> entry:page.candidates.entrySet())
81 if(registry.reviewLedger().get(contract,scope(entry.getValue()),identity(entry.getValue()))==null&&!inFlight.contains(entry.getKey()))pending.put(entry.getKey(),entry.getValue());
82@@ -154,7 +163,6 @@ public final class JevSemanticReview implements AutoCloseable {
83 requestPage=new Page(page.cacheKey+"|pending",pendingPayload(page.payload,fingerprintMisses),fingerprintMisses);
84 request=JevSemanticProtocol.request("semantic-"+sequence.incrementAndGet(),model,requestPage.payload,maxBytes);
85 }
86- int retries=0;
87 while(true) {
88 long remaining=deadline-System.nanoTime();
89 if(synchronous && (remaining<=0 || Thread.currentThread().isInterrupted())) {
90@@ -162,12 +170,17 @@ public final class JevSemanticReview implements AutoCloseable {
91 }
92 JevDeadline.check();
93 int attempt=nextAttempt(request.inputFingerprint);
94+ // Capture before submitting, so expiry cannot orphan an accepted call.
95+ String sourceGui=JevEvidence.guiIdentity(tree);
96+ long historyRevision=registry.executionRevision(state);
97 receipt("semantic-request-"+request.requestId,requestReceipt(request,requestPage.candidates,attempt));
98 registry.semanticEvent("JEV_REQUEST_SENT",currentStep,state.getGraphId(),null,JevRegistry.fields(
99 "request_id",request.requestId,"candidate_count",requestPage.candidates.size(),"wait_mode",synchronous?"synchronous":"asynchronous",
100 "attempt",attempt,"input_fingerprint",request.inputFingerprint));
101 JevDeadline.check();
102- 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));
103+ JevTransport.Call call=synchronous
104+ ?transport.submitRetained(request.requestId,request.bodyJson,requestLifetimeMs,Math.max(16384,maxBytes))
105+ :transport.submit(request.requestId,request.bodyJson,timeoutMs,Math.max(16384,maxBytes));
106 if(call==null) {
107 requestFailure(request,attempt,"QUEUE_FULL",0,null);
108 if(retries++<MAX_RETRIES){
109@@ -175,16 +188,18 @@ public final class JevSemanticReview implements AutoCloseable {
110 }
111 failOpen(state,"QUEUE_FULL");return;
112 }
113- String sourceGui=JevEvidence.guiIdentity(tree);
114 Cache submitted=new Cache(requestPage.candidates,request,call,attempt,sourceGui);
115+ submitted.sourceVersion=modelVersion;submitted.historyRevision=historyRevision;
116 pages.put(requestPage.cacheKey,submitted);
117+ if(synchronous)archiveOnArrival(submitted,currentStep);
118 inFlight.addAll(requestPage.candidates.keySet());
119 for(Candidate candidate:pending.values())registry.reviewLedger().bind(contract,scope(candidate),identity(candidate),sourceGui);
120 if(!synchronous)return;
121 registry.semanticEvent("SEMANTIC_WAIT_STARTED",currentStep,state.getGraphId(),null,JevRegistry.fields(
122 "request_id",request.requestId,"candidate_count",requestPage.candidates.size(),"attempt",attempt));
123- JevTransport.Result response=call.await();
124+ JevTransport.Result response=call.awaitUntil(deadline);
125 JevDeadline.check();
126+ if(response==null)throw new JevDeadline.Expired();
127 boolean completed=consumeCache(requestPage.cacheKey,submitted,response);
128 if(completed)return;
129 if(retries++>=MAX_RETRIES){failOpen(state,"RETRIES_EXHAUSTED");return;}
130@@ -200,6 +215,74 @@ public final class JevSemanticReview implements AutoCloseable {
131 registry.semanticEvent("SEMANTIC_FAIL_OPEN",currentStep,state.getGraphId(),null,
132 JevRegistry.fields("reason",reason,"fallback","native","semantic_selection_blocked",true));
133 }
134+ private void retain(Cache cache) {
135+ retained.put(cache.request.requestId,cache);
136+ while(retained.size()>retainedCapacity) {
137+ Iterator<Map.Entry<String,Cache>> it=retained.entrySet().iterator();
138+ Cache evicted=it.next().getValue();it.remove();evicted.call.cancel("RETAINED_CAPACITY");
139+ }
140+ }
141+ /** Archive raw bytes even if APE never revisits. No decisions are parsed or applied here. */
142+ private void archiveOnArrival(Cache cache,final long sourceStep) {
143+ if(receiptSink==null)return;
144+ final JevSemanticProtocol.Request request=cache.request;
145+ final long sourceVersion=cache.sourceVersion,historyRevision=cache.historyRevision;
146+ final String sourceGui=cache.sourceGui;
147+ cache.call.onComplete(new JevTransport.Completion(){public void completed(final JevTransport.Result result) {
148+ try {receiptWorker.execute(new Runnable(){public void run() {
149+ try {
150+ JSONObject value=new JSONObject().put("schema",1).put("kind","semantic_provider_arrival")
151+ .put("request_id",request.requestId).put("endpoint",endpoint).put("model",model)
152+ .put("contract",contract).put("source_step",sourceStep).put("source_version",sourceVersion)
153+ .put("history_revision",historyRevision).put("source_gui",sourceGui)
154+ .put("input_signature",request.inputSignature).put("input_fingerprint",request.inputFingerprint)
155+ .put("http_status",result.httpStatus).put("error",result.error==null?JSONObject.NULL:result.error)
156+ .put("raw_response",result.raw==null?JSONObject.NULL:result.raw)
157+ .put("submitted_ns",result.submittedNanos).put("sent_ns",result.sentNanos)
158+ .put("completed_ns",result.completedNanos).put("wait_ns",result.waitNanos())
159+ .put("validated",false).put("applied",false);
160+ if(!receiptSink.save("semantic-arrival-"+request.requestId,value))healthy=false;
161+ }catch(JSONException e){healthy=false;}catch(RuntimeException e){healthy=false;}
162+ }});}catch(RejectedExecutionException e){healthy=false;}
163+ }});
164+ }
165+ /** A later review may reuse raw work only against the same visible input and history. */
166+ private int resumeRetained(State state,GUITree tree,Page page,long deadline) {
167+ for(Map.Entry<String,Cache> item:new ArrayList<Map.Entry<String,Cache>>(retained.entrySet())) {
168+ JevDeadline.check();Cache old=item.getValue();
169+ if(old.sourceVersion!=modelVersion) {
170+ old.call.cancel("STALE_HISTORY_OR_VERSION");retained.remove(item.getKey());continue;
171+ }
172+ JevTransport.Result response=old.call.poll();
173+ if(!old.request.inputSignature.equals(page.payload.optString("input_signature"))) {
174+ if(response!=null && response.error!=null)retained.remove(item.getKey());
175+ continue;
176+ }
177+ if(old.historyRevision!=registry.executionRevision(state)) {
178+ old.call.cancel("STALE_HISTORY_OR_VERSION");retained.remove(item.getKey());continue;
179+ }
180+ if(!old.sourceGui.equals(JevEvidence.guiIdentity(tree)))continue;
181+ Map<String,Candidate> current=new LinkedHashMap<String,Candidate>();
182+ for(String key:old.candidates.keySet()) {
183+ Candidate c=page.candidates.get(key);
184+ if(c!=null && old.byId.containsKey(c.candidateId) && old.byId.get(c.candidateId).key.equals(key))current.put(key,c);
185+ }
186+ if(current.size()!=old.request.candidateIds.size()) {
187+ old.call.cancel("STALE_CANDIDATES");retained.remove(item.getKey());continue;
188+ }
189+ Cache rebound=new Cache(current,old.request,old.call,old.attempt,JevEvidence.guiIdentity(tree));
190+ rebound.sourceVersion=old.sourceVersion;rebound.historyRevision=old.historyRevision;
191+ // Keep it recoverable even if parsing/receipt processing exhausts this later budget.
192+ pages.put(item.getKey(),rebound);retained.remove(item.getKey());
193+ if(response==null)response=old.call.awaitUntil(deadline);
194+ JevDeadline.check();
195+ if(response==null)throw new JevDeadline.Expired();
196+ if(consumeCache(item.getKey(),rebound,response))return -1;
197+ // Protocol / HTTP failures remain failures, rather than reusable semantic answers.
198+ return 1;
199+ }
200+ return 0;
201+ }
202 private static final class Page {
203 final String cacheKey;final JSONObject payload;final Map<String,Candidate> candidates;
204 Page(String k,JSONObject p,Map<String,Candidate> c){cacheKey=k;payload=p;candidates=c;}
205@@ -442,6 +525,7 @@ public final class JevSemanticReview implements AutoCloseable {
206 return new JSONObject().put("schema",1).put("kind",synchronous?"sync_semantic_request":"async_semantic_request").put("request_id",request.requestId)
207 .put("attempt",attempt).put("endpoint",endpoint).put("model",model).put("contract",contract).put("step",currentStep)
208 .put("retry_count",attempt-1)
209+ .put("request_lifetime_ms",synchronous?requestLifetimeMs:timeoutMs).put("late_response_reuse",synchronous)
210 .put("input_signature",request.inputSignature).put("input_fingerprint",request.inputFingerprint).put("candidate_ids",ids)
211 .put("request_bytes",request.requestBytes).put("timeout_ms",timeoutMs).put("body",new JSONObject(request.bodyJson));
212 } catch(JSONException error) { return new JSONObject(); }
213@@ -475,5 +559,5 @@ public final class JevSemanticReview implements AutoCloseable {
214 "scrollable",node.isScrollable(),"password",node.isPassword(),"input_text",JevEvidence.boundedText(node.getInputText()),
215 "bounds",new JSONArray().put(b.left).put(b.top).put(b.right).put(b.bottom),"role_signature",JevEvidence.roleSignature(node));
216 }
217- public void close(){reviewLock.lock();try {receiptWorker.shutdownNow();transport.close();pages.clear();inFlight.clear();served.clear();} finally {reviewLock.unlock();}}
218+ public void close(){reviewLock.lock();try {receiptWorker.shutdownNow();transport.close();pages.clear();retained.clear();inFlight.clear();served.clear();} finally {reviewLock.unlock();}}
219 }
220diff --git a/native/src/com/android/commands/monkey/ape/jev/JevTransport.java b/native/src/com/android/commands/monkey/ape/jev/JevTransport.java
221index f35486c..beed490 100644
222--- a/native/src/com/android/commands/monkey/ape/jev/JevTransport.java
223+++ b/native/src/com/android/commands/monkey/ape/jev/JevTransport.java
224@@ -13,6 +13,7 @@ import java.util.LinkedHashMap;
225
226 /** One bounded HTTP worker. Deadlines include the queue and never wait for socket cleanup. */
227 public final class JevTransport implements Closeable {
228+ interface Completion { void completed(Result result); }
229 public static final class Result {
230 public final String raw,error;
231 public final int httpStatus;
232@@ -33,16 +34,26 @@ public final class JevTransport implements Closeable {
233 private volatile HttpURLConnection connection;
234 private final java.util.concurrent.atomic.AtomicReference<Result> result=new java.util.concurrent.atomic.AtomicReference<Result>();
235 private final java.util.concurrent.atomic.AtomicBoolean consumed=new java.util.concurrent.atomic.AtomicBoolean();
236+ private final java.util.concurrent.atomic.AtomicBoolean notified=new java.util.concurrent.atomic.AtomicBoolean();
237+ private volatile Completion completion;
238 private final java.util.concurrent.CountDownLatch ready=new java.util.concurrent.CountDownLatch(1);
239 private TimerTask expiry;
240- private Call(String id,String body,int timeout,int limit) {
241+ private Call(String id,String body,int timeout,int limit,boolean retainAfterWait) {
242 requestId=id;this.body=body.getBytes(StandardCharsets.UTF_8);responseLimit=limit;
243- deadline=Math.min(JevDeadline.absolute(),submitted+TimeUnit.MILLISECONDS.toNanos(timeout));
244+ long requestDeadline=submitted+TimeUnit.MILLISECONDS.toNanos(timeout);
245+ deadline=retainAfterWait?requestDeadline:Math.min(JevDeadline.absolute(),requestDeadline);
246 }
247 private boolean finish(String raw,String error,int status) {
248 if(!result.compareAndSet(null,new Result(this,raw,error,status)))return false;
249 ready.countDown();
250- if(expiry!=null)expiry.cancel();return true;
251+ if(expiry!=null)expiry.cancel();notifyCompletion();return true;
252+ }
253+ // The listener only enqueues immutable receipt data; it must never wait or touch GUI state.
254+ void onComplete(Completion listener) {completion=listener;notifyCompletion();}
255+ private void notifyCompletion() {
256+ Completion listener=completion;Result value=result.get();
257+ if(listener!=null && value!=null && notified.compareAndSet(false,true))
258+ listener.completed(value);
259 }
260 public Result poll() {
261 if(result.get()==null && System.nanoTime()>=deadline)cancel("TIMEOUT");
262@@ -59,6 +70,13 @@ public final class JevTransport implements Closeable {
263 } catch(InterruptedException interrupted){cancel("INTERRUPTED");Thread.currentThread().interrupt();}
264 return result.get();
265 }
266+ /** End the caller's wait without cancelling an already submitted request. */
267+ public Result awaitUntil(long waitDeadline) {
268+ if(poll()==null)try {
269+ ready.await(Math.max(0,Math.min(waitDeadline,deadline)-System.nanoTime()),TimeUnit.NANOSECONDS);
270+ }catch(InterruptedException e){Thread.currentThread().interrupt();}
271+ return poll();
272+ }
273 public void cancel(String reason) {
274 if(finish(null,reason==null?"CANCELLED":reason,0)) {
275 waiting.remove(this);
276@@ -153,12 +171,19 @@ public final class JevTransport implements Closeable {
277 catch(RuntimeException ignored){ }
278 }},"ape-"+requestClass+"-cleanup");cleanup.setDaemon(true);cleanup.start();
279 }
280- public Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes) {JevDeadline.lock(submitLock);try {
281+ public Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes) {
282+ return submit(requestId,body,timeoutMs,maxResponseBytes,false);
283+ }
284+ /** Only the network lifetime outlives the caller's budget; enqueue remains budgeted. */
285+ Call submitRetained(String requestId,String body,int timeoutMs,int maxResponseBytes) {
286+ return submit(requestId,body,timeoutMs,maxResponseBytes,true);
287+ }
288+ private Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes,boolean retained) {JevDeadline.lock(submitLock);try {
289 if(requestId==null || requestId.isEmpty() || body==null || timeoutMs<=0 || maxResponseBytes<=0)
290 throw new IllegalArgumentException("invalid request bounds");
291 if(closed)return null;
292 JevDeadline.check();used=true;
293- final Call call=new Call(requestId,body,timeoutMs,maxResponseBytes);
294+ final Call call=new Call(requestId,body,timeoutMs,maxResponseBytes,retained);
295 call.expiry=new TimerTask(){public void run(){call.cancel("TIMEOUT");}};
296 // Schedule before publishing: an immediate response may cancel this task.
297 deadlines.schedule(call.expiry,Math.max(1,TimeUnit.NANOSECONDS.toMillis(Math.max(0,call.deadline-System.nanoTime()))));