Move Android bridge traffic onto AIDL

Add a shared Binder contract for the Agent/Genie control plane and switch the Genie runtime to use the exported Agent bridge service for runtime status and proxied HTTP, leaving AgentSDK questions only for real Genie dialogue.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Iliyan Malchev
2026-03-19 09:22:04 -07:00
parent 1340af08aa
commit 30ae7c28e1
22 changed files with 824 additions and 424 deletions

View File

@@ -11,8 +11,6 @@ import java.util.concurrent.Executor
class AgentSessionController(context: Context) {
companion object {
private const val PREFERRED_GENIE_PACKAGE = "com.openai.codex.genie"
private const val BRIDGE_REQUEST_PREFIX = "__codex_bridge__ "
private const val BRIDGE_RESPONSE_PREFIX = "__codex_bridge_result__ "
}
private val agentManager = context.getSystemService(AgentManager::class.java)
@@ -222,7 +220,7 @@ class AgentSessionController(context: Context) {
private fun findLastEventMessage(events: List<AgentSessionEvent>, type: Int): String? {
for (index in events.indices.reversed()) {
val event = events[index]
if (event.type == type && event.message != null && !isInternalBridgeEvent(event)) {
if (event.type == type && event.message != null) {
return event.message
}
}
@@ -230,24 +228,14 @@ class AgentSessionController(context: Context) {
}
private fun renderTimeline(events: List<AgentSessionEvent>): String {
val visibleEvents = events.filterNot(::isInternalBridgeEvent)
if (visibleEvents.isEmpty()) {
if (events.isEmpty()) {
return "No framework events yet."
}
return visibleEvents.joinToString("\n") { event ->
return events.joinToString("\n") { event ->
"${eventTypeToString(event.type)}: ${event.message ?: ""}"
}
}
private fun isInternalBridgeEvent(event: AgentSessionEvent): Boolean {
val message = event.message ?: return false
return when (event.type) {
AgentSessionEvent.TYPE_QUESTION -> message.startsWith(BRIDGE_REQUEST_PREFIX)
AgentSessionEvent.TYPE_ANSWER -> message.startsWith(BRIDGE_RESPONSE_PREFIX)
else -> false
}
}
private fun eventTypeToString(type: Int): String {
return when (type) {
AgentSessionEvent.TYPE_TRACE -> "Trace"

View File

@@ -0,0 +1,64 @@
package com.openai.codexd
import android.app.Service
import android.content.Intent
import android.os.IBinder
import android.util.Log
import com.openai.codex.bridge.BridgeHttpRequest
import com.openai.codex.bridge.BridgeHttpResponse
import com.openai.codex.bridge.BridgeRuntimeStatus
import com.openai.codex.bridge.ICodexAgentBridgeService
import java.io.IOException
class CodexAgentBridgeService : Service() {
companion object {
const val PERMISSION_BIND_AGENT_BRIDGE = "com.openai.codex.permission.BIND_AGENT_BRIDGE"
private const val TAG = "CodexAgentBridgeSvc"
}
private val binder = object : ICodexAgentBridgeService.Stub() {
override fun getRuntimeStatus(): BridgeRuntimeStatus {
val status = runCatching {
CodexdLocalClient.waitForRuntimeStatus(this@CodexAgentBridgeService)
}.getOrElse { err ->
throw err.asBinderError("getRuntimeStatus")
}
Log.i(TAG, "Served runtime status")
return BridgeRuntimeStatus(
status.authenticated,
status.accountEmail,
status.clientCount,
status.modelProviderId,
status.configuredModel,
status.effectiveModel,
status.upstreamBaseUrl,
)
}
override fun sendHttpRequest(request: BridgeHttpRequest): BridgeHttpResponse {
val response = runCatching {
CodexdLocalClient.waitForResponse(
this@CodexAgentBridgeService,
request.method,
request.path,
request.body,
)
}.getOrElse { err ->
throw err.asBinderError("sendHttpRequest")
}
Log.i(TAG, "Proxied ${request.method} ${request.path}")
return BridgeHttpResponse(response.statusCode, response.body)
}
}
override fun onBind(intent: Intent?): IBinder {
return binder
}
private fun Throwable.asBinderError(operation: String): IllegalStateException {
val detail = message ?: javaClass.simpleName
val message = "$operation failed: $detail"
Log.w(TAG, message, this)
return IllegalStateException(message, this)
}
}

View File

@@ -9,7 +9,6 @@ import android.util.Log
import org.json.JSONArray
import org.json.JSONObject
import java.io.IOException
import java.util.concurrent.ConcurrentHashMap
import kotlin.concurrent.thread
class CodexAgentService : AgentService() {
@@ -17,23 +16,17 @@ class CodexAgentService : AgentService() {
private const val TAG = "CodexAgentService"
private const val BRIDGE_ANSWER_RETRY_COUNT = 10
private const val BRIDGE_ANSWER_RETRY_DELAY_MS = 50L
private const val BRIDGE_REQUEST_PREFIX = "__codex_bridge__ "
private const val BRIDGE_RESPONSE_PREFIX = "__codex_bridge_result__ "
private const val METHOD_GET_AUTH_STATUS = "get_auth_status"
private const val METHOD_HTTP_REQUEST = "http_request"
private const val AUTO_ANSWER_INSTRUCTIONS =
"You are Codex acting as the Android Agent supervising a Genie execution. Reply with the exact free-form answer that should be sent back to the Genie. Keep it short and actionable. If the Genie can proceed without extra constraints, reply with exactly: continue"
private const val MAX_AUTO_ANSWER_CONTEXT_CHARS = 800
}
private val handledBridgeRequests = ConcurrentHashMap.newKeySet<String>()
private val handledGenieQuestions = ConcurrentHashMap.newKeySet<String>()
private val pendingGenieQuestions = ConcurrentHashMap.newKeySet<String>()
private val handledGenieQuestions = java.util.concurrent.ConcurrentHashMap.newKeySet<String>()
private val pendingGenieQuestions = java.util.concurrent.ConcurrentHashMap.newKeySet<String>()
private val agentManager by lazy { getSystemService(AgentManager::class.java) }
override fun onSessionChanged(session: AgentSessionInfo) {
Log.i(TAG, "onSessionChanged $session")
handleInternalBridgeQuestion(session.sessionId)
maybeAutoAnswerGenieQuestion(session)
updateQuestionNotification(session)
}
@@ -45,94 +38,6 @@ class CodexAgentService : AgentService() {
pendingGenieQuestions.removeIf { it.startsWith("$sessionId:") }
}
private fun handleInternalBridgeQuestion(sessionId: String) {
val manager = agentManager ?: return
val events = manager.getSessionEvents(sessionId)
val question = events.lastOrNull { event ->
event.type == AgentSessionEvent.TYPE_QUESTION && event.message != null
}?.message ?: return
if (!question.startsWith(BRIDGE_REQUEST_PREFIX)) {
return
}
val requestJson = runCatching {
JSONObject(question.removePrefix(BRIDGE_REQUEST_PREFIX))
}.getOrElse { err ->
Log.w(TAG, "Ignoring malformed bridge question for $sessionId", err)
return
}
val requestId = requestJson.optString("requestId")
val method = requestJson.optString("method")
if (requestId.isBlank() || method.isBlank()) {
return
}
val requestKey = "$sessionId:$requestId"
if (hasAnswerForRequest(events, requestId) || !handledBridgeRequests.add(requestKey)) {
return
}
thread(name = "CodexAgentBridge-$requestId") {
val response = when (method) {
METHOD_GET_AUTH_STATUS -> runCatching { CodexdLocalClient.waitForAuthStatus(this) }
.fold(
onSuccess = { status ->
JSONObject()
.put("requestId", requestId)
.put("ok", true)
.put("authenticated", status.authenticated)
.put("accountEmail", status.accountEmail)
.put("clientCount", status.clientCount)
},
onFailure = { err ->
JSONObject()
.put("requestId", requestId)
.put("ok", false)
.put("error", err.message ?: err::class.java.simpleName)
},
)
METHOD_HTTP_REQUEST -> {
val httpMethod = requestJson.optString("httpMethod")
val path = requestJson.optString("path")
val body = if (requestJson.isNull("body")) null else requestJson.optString("body")
if (httpMethod.isBlank() || path.isBlank()) {
JSONObject()
.put("requestId", requestId)
.put("ok", false)
.put("error", "Missing httpMethod or path")
} else {
runCatching {
CodexdLocalClient.waitForResponse(this, httpMethod, path, body)
}.fold(
onSuccess = { httpResponse ->
JSONObject()
.put("requestId", requestId)
.put("ok", true)
.put("statusCode", httpResponse.statusCode)
.put("body", httpResponse.body)
},
onFailure = { err ->
JSONObject()
.put("requestId", requestId)
.put("ok", false)
.put("error", err.message ?: err::class.java.simpleName)
},
)
}
}
else -> JSONObject()
.put("requestId", requestId)
.put("ok", false)
.put("error", "Unknown bridge method: $method")
}
runCatching {
answerQuestionWithRetry(manager, sessionId, "$BRIDGE_RESPONSE_PREFIX$response")
}.onFailure { err ->
handledBridgeRequests.remove(requestKey)
Log.w(TAG, "Failed to answer bridge question for $sessionId", err)
}
}
}
private fun answerQuestionWithRetry(manager: AgentManager, sessionId: String, response: String) {
repeat(BRIDGE_ANSWER_RETRY_COUNT) { attempt ->
runCatching {
@@ -148,21 +53,6 @@ class CodexAgentService : AgentService() {
}
}
private fun hasAnswerForRequest(events: List<AgentSessionEvent>, requestId: String): Boolean {
return events.any { event ->
if (event.type != AgentSessionEvent.TYPE_ANSWER || event.message == null) {
return@any false
}
val message = event.message
if (!message.startsWith(BRIDGE_RESPONSE_PREFIX)) {
return@any false
}
runCatching {
JSONObject(message.removePrefix(BRIDGE_RESPONSE_PREFIX)).optString("requestId")
}.getOrNull() == requestId
}
}
private fun isSessionWaitingForUser(manager: AgentManager, sessionId: String): Boolean {
return manager.getSessions(Process.myUid() / 100000).any { session ->
session.sessionId == sessionId &&
@@ -248,7 +138,7 @@ class CodexAgentService : AgentService() {
val requestBody = JSONObject()
.put("model", model)
.put("store", false)
.put("stream", false)
.put("stream", true)
.put("instructions", AUTO_ANSWER_INSTRUCTIONS)
.put(
"input",
@@ -290,7 +180,6 @@ class CodexAgentService : AgentService() {
private fun renderRecentContext(events: List<AgentSessionEvent>): String {
val context = events
.filterNot(::isInternalBridgeEvent)
.takeLast(6)
.joinToString("\n") { event ->
"${eventTypeToString(event.type)}: ${event.message ?: ""}"
@@ -302,7 +191,11 @@ class CodexAgentService : AgentService() {
}
private fun parseResponsesOutputText(body: String): String {
val data = JSONObject(body)
val trimmedBody = body.trim()
if (trimmedBody.startsWith("event:") || trimmedBody.startsWith("data:")) {
return parseResponsesStreamOutputText(trimmedBody)
}
val data = JSONObject(trimmedBody)
val directOutput = data.optString("output_text")
if (directOutput.isNotBlank()) {
return directOutput
@@ -327,23 +220,59 @@ class CodexAgentService : AgentService() {
return combined
}
private fun parseResponsesStreamOutputText(body: String): String {
val deltaText = StringBuilder()
val completedItems = mutableListOf<String>()
body.split("\n\n").forEach { rawEvent ->
val lines = rawEvent.lineSequence().map(String::trimEnd).toList()
if (lines.isEmpty()) {
return@forEach
}
val dataPayload = lines
.filter { it.startsWith("data:") }
.joinToString("\n") { it.removePrefix("data:").trimStart() }
.trim()
if (dataPayload.isEmpty() || dataPayload == "[DONE]") {
return@forEach
}
val event = JSONObject(dataPayload)
when (event.optString("type")) {
"response.output_text.delta" -> deltaText.append(event.optString("delta"))
"response.output_item.done" -> {
val item = event.optJSONObject("item") ?: return@forEach
val content = item.optJSONArray("content") ?: return@forEach
val text = buildString {
for (index in 0 until content.length()) {
val part = content.optJSONObject(index) ?: continue
if (part.optString("type") == "output_text") {
append(part.optString("text"))
}
}
}
if (text.isNotBlank()) {
completedItems += text
}
}
"response.failed" -> throw IOException(event.toString())
}
}
if (deltaText.isNotBlank()) {
return deltaText.toString()
}
val completedText = completedItems.joinToString("")
if (completedText.isNotBlank()) {
return completedText
}
throw IOException("Responses stream missing output_text content")
}
private fun findVisibleQuestion(events: List<AgentSessionEvent>): String? {
return events.lastOrNull { event ->
event.type == AgentSessionEvent.TYPE_QUESTION &&
!event.message.isNullOrBlank() &&
!isInternalBridgeEvent(event)
!event.message.isNullOrBlank()
}?.message
}
private fun isInternalBridgeEvent(event: AgentSessionEvent): Boolean {
val message = event.message ?: return false
return when (event.type) {
AgentSessionEvent.TYPE_QUESTION -> message.startsWith(BRIDGE_REQUEST_PREFIX)
AgentSessionEvent.TYPE_ANSWER -> message.startsWith(BRIDGE_RESPONSE_PREFIX)
else -> false
}
}
private fun eventTypeToString(type: Int): String {
return when (type) {
AgentSessionEvent.TYPE_TRACE -> "Trace"

View File

@@ -2,11 +2,12 @@ package com.openai.codexd
import android.content.Context
import android.net.LocalSocket
import org.json.JSONObject
import java.io.BufferedInputStream
import java.io.ByteArrayOutputStream
import java.io.File
import java.io.IOException
import java.io.BufferedInputStream
import java.nio.charset.StandardCharsets
import org.json.JSONObject
object CodexdLocalClient {
data class HttpResponse(
@@ -110,23 +111,91 @@ object CodexdLocalClient {
val responseBytes = BufferedInputStream(socket.inputStream).use { it.readBytes() }
socket.close()
val responseText = responseBytes.toString(StandardCharsets.UTF_8)
val splitIndex = responseText.indexOf("\r\n\r\n")
val splitIndex = responseBytes.indexOfHeaderBodySeparator()
if (splitIndex == -1) {
throw IOException("Invalid HTTP response")
}
val statusLine = responseText.substring(0, splitIndex)
val headerText = responseBytes
.copyOfRange(0, splitIndex)
.toString(StandardCharsets.UTF_8)
val statusLine = headerText
.lineSequence()
.firstOrNull()
.orEmpty()
val statusCode = statusLine.split(" ").getOrNull(1)?.toIntOrNull()
?: throw IOException("Missing status code")
val bodyBytes = responseBytes.copyOfRange(splitIndex + 4, responseBytes.size)
val decodedBodyBytes = if (headerText.contains("Transfer-Encoding: chunked", ignoreCase = true)) {
decodeChunkedBody(bodyBytes)
} else {
bodyBytes
}
return HttpResponse(
statusCode = statusCode,
body = responseText.substring(splitIndex + 4),
body = decodedBodyBytes.toString(StandardCharsets.UTF_8),
)
}
private fun ByteArray.indexOfHeaderBodySeparator(): Int {
for (index in 0 until size - 3) {
if (
this[index] == '\r'.code.toByte() &&
this[index + 1] == '\n'.code.toByte() &&
this[index + 2] == '\r'.code.toByte() &&
this[index + 3] == '\n'.code.toByte()
) {
return index
}
}
return -1
}
private fun decodeChunkedBody(bodyBytes: ByteArray): ByteArray {
val output = ByteArrayOutputStream(bodyBytes.size)
var cursor = 0
while (cursor < bodyBytes.size) {
val lineEnd = bodyBytes.indexOfCrlf(cursor)
if (lineEnd == -1) {
throw IOException("Invalid chunked response")
}
val sizeLine = bodyBytes
.copyOfRange(cursor, lineEnd)
.toString(StandardCharsets.US_ASCII)
.substringBefore(';')
.trim()
val chunkSize = sizeLine.toIntOrNull(radix = 16)
?: throw IOException("Invalid chunk size: $sizeLine")
cursor = lineEnd + 2
if (chunkSize == 0) {
break
}
val nextCursor = cursor + chunkSize
if (nextCursor > bodyBytes.size) {
throw IOException("Chunk exceeds body length")
}
output.write(bodyBytes, cursor, chunkSize)
cursor = nextCursor
if (
cursor + 1 >= bodyBytes.size ||
bodyBytes[cursor] != '\r'.code.toByte() ||
bodyBytes[cursor + 1] != '\n'.code.toByte()
) {
throw IOException("Invalid chunk terminator")
}
cursor += 2
}
return output.toByteArray()
}
private fun ByteArray.indexOfCrlf(startIndex: Int): Int {
for (index in startIndex until size - 1) {
if (this[index] == '\r'.code.toByte() && this[index + 1] == '\n'.code.toByte()) {
return index
}
}
return -1
}
private fun parseAuthStatus(body: String): AuthStatus {
val json = JSONObject(body)
val accountEmail =