Prepare an agent whose native release inspection was blocked

workspaces/runtime/packages/runtime-services/src/account_transfer_executor.rs

@@ -1964,8 +1964,10 @@ mod retained_executor_tests {
         store
             .transaction_with_error(|tx| {
                 let mut row = AgentRepository::get(tx, &AgentId::from("lead"))?.unwrap();
-                row.extra
-                    .insert("nativeRelease".into(), json!({"phase":"pending"}));
+                row.extra.insert(
+                    "nativeRelease".into(),
+                    json!({"phase":"unknown","resetPending":true}),
+                );
                 AgentRepository::put(tx, &row)
             })
             .await

workspaces/runtime/packages/runtime-services/src/account_worker/preparation/snapshot.rs

@@ -794,6 +794,34 @@ pub(super) fn context_repair_blocked(record: &Value) -> bool {
             Some("preparing" | "submitted" | "unknown" | "ready")
         )
 }
+/// Whether a saved native release keeps this agent's queued input out of
+/// thread preparation. Python `Runtime.prepare_locked` refuses a pending tool
+/// reset (`resetPending`), then `codex_native_release.reconcile_unknown` acts
+/// only on an unsettled `unknown`, `checking` or `unsubscribing` release: a
+/// captured reset waits for exact native closure, and an unsettled release on
+/// this thread and connection retries `thread/unsubscribe`. This adapter has no
+/// native release driver, so it holds in both cases. Every other release
+/// prepares normally, including `blocked`, where the inspection found native
+/// work and never unsubscribed, and an unsettled release on an earlier
+/// connection, which Python also ignores.
+pub(super) fn native_release_held(record: &Value, connection: &str) -> bool {
+    let release = &record["nativeRelease"];
+    if nonempty(&release["resetPending"]) {
+        return true;
+    }
+    let phase = release["phase"].as_str();
+    if !matches!(phase, Some("unknown" | "checking" | "unsubscribing")) {
+        return false;
+    }
+    if release.get("resetActorEpoch").is_some() {
+        return true;
+    }
+    if phase == Some("checking") && nonempty(&release["inspectionPhase"]) {
+        return false;
+    }
+    release["threadId"] == record["threadId"]
+        && release["connectionId"].as_str() == Some(connection)
+}
 pub(super) fn nonempty(value: &Value) -> bool {
     match value {
         Value::Null => false,
@@ -886,9 +914,7 @@ pub(super) fn thread_snapshot(
         Some(thread::HoldReason::AccountTransfer)
     } else if context_repair_blocked(record) {
         Some(thread::HoldReason::ContextRepair)
-    } else if record.get("nativeRelease").is_some_and(|v| {
-        !v.is_null() && !matches!(v["phase"].as_str(), Some("resumed" | "released"))
-    }) {
+    } else if native_release_held(record, connection) {
         Some(thread::HoldReason::NativeRelease)
     } else {
         None

workspaces/runtime/packages/runtime-services/src/account_worker/tests/preparation_review.rs

@@ -191,6 +191,234 @@ async fn context_repair_phases_and_complaint_ownership_match_python() {
     }
 }
 
+/// Real Python `Runtime.prepare_locked` and `reconcile_unknown` over the same
+/// saved row, with only the runtime host replaced. A release Python ignores
+/// reaches the next preparation call (`proceed`); a raise keeps the input
+/// queued (`hold`); `unsubscribe` is the native release retry that the Rust
+/// adapter holds for instead.
+const NATIVE_RELEASE_ORACLE: &str = r#"
+import contextlib, json, sys, threading
+sys.path.insert(0, sys.argv[1])
+from oracle_guard import install_guard
+_GUARD = install_guard(block_processes=True)
+sys.path.insert(0, sys.argv[2])
+import codex_context_repair, codex_native_tools
+from codex_runtime import Runtime
+
+class Proceed(Exception):
+    pass
+
+class Preparations(dict):
+    def get(self, *args):
+        raise Proceed
+
+class Server:
+    def __init__(self):
+        self.methods = []
+    def call(self, method, params, timeout=None):
+        self.methods.append(method)
+        return {"status": "unsubscribed"}
+
+class Host:
+    def __init__(self, record, connection):
+        self.record, self.connection = record, connection
+        self.lock, self.closed, self.loaded = threading.RLock(), False, set()
+        self.preparations, self.server = Preparations(), Server()
+    @contextlib.contextmanager
+    def db(self):
+        yield None
+    def agent(self, agent_id, db=None):
+        return json.loads(json.dumps(self.record))
+    def put(self, db, table, record):
+        self.record = record
+    def agent_connection(self, agent):
+        return self.connection
+    def server_for(self, account, connection):
+        return self.server
+    def connect_agent(self, agent):
+        return self.server
+
+codex_context_repair.assert_context_available = lambda agent: None
+codex_native_tools.account_reserved = lambda runtime, account: False
+outcomes = []
+for case in json.loads(sys.argv[3]):
+    host = Host(case["agent"], case["connection"])
+    try:
+        Runtime.prepare_locked(host, host.agent(None))
+        outcomes.append("returned")
+    except Proceed:
+        outcomes.append("unsubscribe" if host.server.methods else "proceed")
+    except Exception:
+        outcomes.append("hold")
+print(json.dumps(outcomes))
+"#;
+
+fn python_native_release_outcomes(cases: &Value) -> Vec<String> {
+    let manifest = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    let source = manifest.join("../../apps/server/src");
+    let output = Command::new("timeout")
+        .arg("120")
+        .arg(crate::agents_create::prepared_python(&source))
+        .arg("-B")
+        .arg("-c")
+        .arg(NATIVE_RELEASE_ORACLE)
+        .arg(manifest.join("tests/turn_lifecycle"))
+        .arg(&source)
+        .arg(cases.to_string())
+        .output()
+        .unwrap();
+    assert!(
+        output.status.success(),
+        "{}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    serde_json::from_slice(&output.stdout).unwrap()
+}
+
+/// A release inspection that found native work (`blocked`) never unsubscribed,
+/// so the thread stayed loaded. Python prepares such an agent normally; only a
+/// pending tool reset or an unsettled release on the current thread and
+/// connection keeps the input queued.
+#[tokio::test]
+async fn native_release_phases_match_python() {
+    // Turn preparation names its connection by the provider generation.
+    const GENERATION: u64 = 1;
+    const CONNECTION: &str = "1";
+    let probe = State::new().await;
+    let probe_id = probe.agent.clone();
+    let base = probe
+        .bridge
+        .run(move |store| async move {
+            let db = store.read()?;
+            Ok::<_, WorkerError>(
+                AgentRepository::get(&db, &AgentId::from(probe_id.as_str()))?
+                    .unwrap()
+                    .value(),
+            )
+        })
+        .await
+        .unwrap();
+    let thread = base["threadId"].clone();
+    let release = |connection: &str, phase: Value| {
+        json!({"id":"release","phase":phase,"threadId":thread,"connectionId":connection,
+            "accountKey":base["accountKey"],"targetEpoch":base["epoch"],
+            "targetRootId":base["rootId"],"targetParentId":base["parentId"],
+            "at":100.0,"resetPending":false,
+            "supervisorIdentity":{"generation":1,"handle":"account:default"}})
+    };
+    let with = |mut value: Value, fields: Value| {
+        for (key, field) in fields.as_object().unwrap() {
+            value[key] = field.clone();
+        }
+        value
+    };
+    let cases = [
+        // The live record: an idle-release inspection blocked by a background
+        // command on a connection from an earlier backend.
+        (
+            "blocked-earlier-connection",
+            with(
+                release("earlier-connection", json!("blocked")),
+                json!({"error":"native background command is active"}),
+            ),
+        ),
+        (
+            "blocked-current-connection",
+            release(CONNECTION, json!("blocked")),
+        ),
+        (
+            "superseded-inspection",
+            with(
+                release(CONNECTION, Value::Null),
+                json!({"inspectionPhase":"superseded"}),
+            ),
+        ),
+        ("released", release(CONNECTION, json!("released"))),
+        ("resumed", release(CONNECTION, json!("resumed"))),
+        ("unrecognized-phase", release(CONNECTION, json!("pending"))),
+        (
+            "reset-pending",
+            with(
+                release(CONNECTION, json!("released")),
+                json!({"resetPending":true}),
+            ),
+        ),
+        (
+            "unsubscribing-current",
+            release(CONNECTION, json!("unsubscribing")),
+        ),
+        (
+            "unsubscribing-earlier-connection",
+            release("earlier-connection", json!("unsubscribing")),
+        ),
+        ("unknown-current", release(CONNECTION, json!("unknown"))),
+        ("checking-current", release(CONNECTION, json!("checking"))),
+        (
+            "checking-inspection",
+            with(
+                release(CONNECTION, json!("checking")),
+                json!({"inspectionPhase":"inspecting"}),
+            ),
+        ),
+        (
+            "captured-reset-without-closure",
+            with(
+                release(CONNECTION, json!("unsubscribing")),
+                json!({"resetActorEpoch":base["epoch"],"submittedAt":101.0}),
+            ),
+        ),
+    ];
+    let python =
+        python_native_release_outcomes(&json!(
+        cases
+            .iter()
+            .map(|(_, release)| json!({"agent":with(base.clone(), json!({"nativeRelease":release})),
+                "connection":CONNECTION}))
+            .collect::<Vec<_>>()
+    ));
+    assert_eq!(python.len(), cases.len());
+    for ((name, release), python) in cases.into_iter().zip(python) {
+        let held = match python.as_str() {
+            "proceed" => false,
+            "hold" | "unsubscribe" => true,
+            other => panic!("{name}: unexpected Python outcome {other}"),
+        };
+        let state = State::new().await;
+        change_agent(&state, "nativeRelease", release).await;
+        let id = state.agent.clone();
+        let epoch = state.epoch;
+        let result = state
+            .bridge
+            .run(move |store| async move {
+                let intent = store::claim(Arc::clone(&store), "default", &id, epoch)
+                    .await?
+                    .unwrap();
+                Ok::<_, WorkerError>(
+                    preparation::prepare_turn_with_connection_id(
+                        Arc::clone(&store),
+                        &test_resources()?,
+                        &intent,
+                        GENERATION,
+                        "transport-connection",
+                    )
+                    .await
+                    .map(|_| ()),
+                )
+            })
+            .await
+            .unwrap();
+        match result {
+            Ok(()) => assert!(!held, "{name}: Python {python} holds, Rust prepared"),
+            Err(WorkerError::PreparationRefused(reason))
+                if reason == "thread-preparation/NativeRelease" =>
+            {
+                assert!(held, "{name}: Python prepares, Rust held")
+            }
+            Err(error) => panic!("{name}: {error}"),
+        }
+    }
+}
+
 pub(super) async fn scenario_late_context_change_refuses_submission() {
     let state = State::new().await;
     let id = state.agent.clone();