Skip to content

fix(assistant): thread-safety in agent run loop and job cancellation #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
faiqq7 wants to merge 3 commits into
base: main
Choose a base branch
from
+31 −14
Open

fix(assistant): thread-safety in agent run loop and job cancellation #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
28d1b48
fix(assistant): thread-safety in agent run loop and job cancellation
faiqq7 Jul 2, 2026
be101d3
Merge upstream main into fix/issue-70-agent-run-loop-thread-safety
faiqq7 Jul 13, 2026
0e08896
fix(assistant): key job cancel flags by long, wire cancel into stop path
faiqq7 Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions ...ackages/apps/OpenPhoneAssistant/src/org/openphone/assistant/AssistantActivityBackend.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@

import java.io.IOException;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicInteger;

public class AssistantActivityBackend extends ComponentActivity {
public interface ComposeStateCallbacks {
Expand Down Expand Up @@ -213,7 +214,7 @@ public void run() {
private volatile boolean mAgentRunCancelled;
private boolean mIslandVoiceLaunch;
private boolean mSuppressNextUserAppend;
private int mAgentRunGeneration;
private final AtomicInteger mAgentRunGeneration = new AtomicInteger();
private int mAuditRefreshGeneration;
private int mVoiceRunGeneration;
private int mChatRunGeneration;
Expand Down Expand Up @@ -1403,7 +1404,7 @@ private void startOpenAiRealtimeVoiceAgent() {
clearVoicePipelineProtection();
mAgentRunCancelled = false;
final int voiceGeneration = ++mVoiceRunGeneration;
final int runGeneration = ++mAgentRunGeneration;
final int runGeneration = mAgentRunGeneration.incrementAndGet();
final OpenAiRealtimeVoiceSession session = new OpenAiRealtimeVoiceSession(endpointConfig,
liveVoiceContinuityContextJson(), "yolo".equals(mAutonomyMode));
mRunningRealtimeVoiceSession = session;
Expand Down Expand Up @@ -1482,7 +1483,7 @@ public String callTool(String toolName, String argumentsJson) {
@Override
public boolean isCancelled() {
return mAgentRunCancelled
|| runGeneration != mAgentRunGeneration
|| runGeneration != mAgentRunGeneration.get()
|| voiceGeneration != mVoiceRunGeneration
|| Thread.currentThread().isInterrupted();
}
Expand Down Expand Up @@ -1567,7 +1568,7 @@ private void startGeminiLiveVoiceAgent() {
clearVoicePipelineProtection();
mAgentRunCancelled = false;
final int voiceGeneration = ++mVoiceRunGeneration;
final int runGeneration = ++mAgentRunGeneration;
final int runGeneration = mAgentRunGeneration.incrementAndGet();
final GeminiLiveVoiceSession session = new GeminiLiveVoiceSession(apiKey,
liveVoiceContinuityContextJson(), "yolo".equals(mAutonomyMode));
mRunningGeminiLiveVoiceSession = session;
Expand Down Expand Up @@ -1646,7 +1647,7 @@ public String callTool(String toolName, String argumentsJson) {
@Override
public boolean isCancelled() {
return mAgentRunCancelled
|| runGeneration != mAgentRunGeneration
|| runGeneration != mAgentRunGeneration.get()
|| voiceGeneration != mVoiceRunGeneration
|| Thread.currentThread().isInterrupted();
}
Expand Down Expand Up @@ -2387,7 +2388,7 @@ private void runOneShotActions(final String originalMessage,
reloadAutonomyMode();
cancelAgentRun();
mAgentRunCancelled = false;
final int runGeneration = ++mAgentRunGeneration;
final int runGeneration = mAgentRunGeneration.incrementAndGet();
final JSONArray proposedActions = decision.proposedActions();
final ModelAdapter adapter = selectedModelAdapter(modelEndpointConfig());
final FrameworkToolExecutor toolExecutor = new FrameworkToolExecutor(this, agentManager);
Expand All @@ -2413,7 +2414,7 @@ public void run() {
adapter.usesCloud(), adapter.privacyDisclosure());
JSONArray results = new JSONArray();
for (int i = 0; i < proposedActions.length(); i++) {
if (mAgentRunCancelled || runGeneration != mAgentRunGeneration) {
if (mAgentRunCancelled || runGeneration != mAgentRunGeneration.get()) {
return;
}
JSONObject action = proposedActions.optJSONObject(i);
Expand Down Expand Up @@ -2481,7 +2482,7 @@ private void postOneShotReply(final int runGeneration, final String reply,
runOnUiThread(new Runnable() {
@Override
public void run() {
if (runGeneration != mAgentRunGeneration) {
if (runGeneration != mAgentRunGeneration.get()) {
return;
}
mAgentThread = null;
Expand All @@ -2505,7 +2506,7 @@ private void showOneShotConfirmation(final int runGeneration, final String taskI
runOnUiThread(new Runnable() {
@Override
public void run() {
if (runGeneration != mAgentRunGeneration) {
if (runGeneration != mAgentRunGeneration.get()) {
try {
mAgentManager.stopTask(taskId, "{\"reason\":\"one_shot_cancelled\"}");
} catch (RuntimeException ignored) {
Expand Down Expand Up @@ -3156,7 +3157,7 @@ private void stopTask() {

private void cancelAgentRun() {
mAgentRunCancelled = true;
mAgentRunGeneration++;
mAgentRunGeneration.incrementAndGet();
ModelAdapter adapter = mRunningModelAdapter;
if (adapter != null) {
adapter.cancel();
Expand Down Expand Up @@ -3226,7 +3227,7 @@ private void runAgent(final String goal) {
reloadAutonomyMode();
cancelAgentRun();
mAgentRunCancelled = false;
final int runGeneration = ++mAgentRunGeneration;
final int runGeneration = mAgentRunGeneration.incrementAndGet();
final String taskId = mActiveTaskId;
final ModelEndpointConfig endpointConfig = modelEndpointConfig();
setTaskText("Working on: " + goal);
Expand Down Expand Up @@ -3288,15 +3289,15 @@ public String callTool(String toolName, String argumentsJson) {
@Override
public boolean isCancelled() {
return mAgentRunCancelled
|| runGeneration != mAgentRunGeneration
|| runGeneration != mAgentRunGeneration.get()
|| Thread.currentThread().isInterrupted();
}
});
trajectory.recordAgentResult(result);
runOnUiThread(new Runnable() {
@Override
public void run() {
if (runGeneration != mAgentRunGeneration) {
if (runGeneration != mAgentRunGeneration.get()) {
return;
}
mAgentThread = null;
Expand Down
1 change: 1 addition & 0 deletions ...ages/apps/OpenPhoneAssistant/src/org/openphone/assistant/agent/FrameworkToolExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2425,6 +2425,7 @@ private String backgroundJobStop(JSONObject arguments) throws JSONException {
return error("bad_background_job_stop");
}
boolean stopped = mAgentJobStore.stop(id);
OpenPhoneAgentJobScheduler.cancelJob(id);
OpenPhoneAgentJobScheduler.scheduleNext(mContext);
return new JSONObject()
.put("status", stopped ? "background_job.stopped"
Expand Down
17 changes: 16 additions & 1 deletion .../apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import org.openphone.assistant.model.OpenAiResponsesAgentAdapter;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

public final class OpenPhoneAgentJobScheduler {
private static final String TAG = "OpenPhoneAgentJobs";
Expand All @@ -32,9 +34,19 @@ public final class OpenPhoneAgentJobScheduler {
private static final long MIN_DELAY_MILLIS = 15_000L;
private static final long STUCK_TIMEOUT_MILLIS = 10L * 60L * 1000L;
private static final int MAX_DUE_PER_CHECK = 3;
private static final ConcurrentHashMap<Long, AtomicBoolean> sCancelFlags =
new ConcurrentHashMap<>();

private OpenPhoneAgentJobScheduler() {}

/** Signals a running background job to stop at its next isCancelled() check. */
public static void cancelJob(long jobId) {
AtomicBoolean flag = sCancelFlags.get(jobId);
if (flag != null) {
flag.set(true);
}
}

public static void checkNow(Context context) {
if (context == null) {
return;
Expand Down Expand Up @@ -111,6 +123,8 @@ private static void runJob(Context context, AgentJobRecord job) {
return;
}
String taskId = null;
AtomicBoolean cancelFlag = sCancelFlags.computeIfAbsent(job.id,
id -> new AtomicBoolean(false));
try {
String response = agentManager.startTask(taskRequestJson(job));
taskId = parseString(response, "task_id");
Expand Down Expand Up @@ -140,7 +154,7 @@ public String callTool(String toolName, String argumentsJson) {

@Override
public boolean isCancelled() {
return false;
return cancelFlag.get() || Thread.currentThread().isInterrupted();
}
});
store.markCompleted(job.id, result, System.currentTimeMillis());
Expand All @@ -150,6 +164,7 @@ public boolean isCancelled() {
} catch (RuntimeException e) {
failJob(context, store, job, "job_error:" + e.getClass().getSimpleName());
} finally {
sCancelFlags.remove(job.id);
if (agentManager != null && taskId != null && !taskId.isEmpty()) {
try {
agentManager.stopTask(taskId, "{\"reason\":\"background_job_finished\"}");
Expand Down