fix(desktop): close the two ways a command still made a same-id twin

The population guard promised the app never adds to the duplicate id
population, and two paths could still walk past it.

`ensure_new_record_is_unique` probed only each other store's live
`sessions/<id>`. A start naming a store whose neighbour holds that id
*archived* was therefore treated as new and opened a record — and the
duplicate appeared later, when the user restored what they had put away.
Another store's archived record is one unarchive away from standing beside
this one, so it counts, and its refusal says which state it found.

`agent.goal` and `agent.compact` never asked at all. Both spawn an engine when
the conversation has no live one, and a spawn writes the record, so either
could file the second twin that `agent.start` refuses to. A goal is reachable
before the first prompt — the composer offers one on a conversation that has
never run — which is exactly when the id is new to its store and old to
another.

Reported by Codex on #886.

Author: tonyfettes

desktop/internal/engine/archive.mbt

@@ -73,6 +73,11 @@ async fn EngineManager::sweep_archived_deletions(self : EngineManager) -> Unit {
 /// not a creation, and passes untouched — including for an id that is
 /// duplicated on disk, because refusing to continue an existing record would
 /// strand the user with a conversation they can open but never answer.
+///
+/// Another store's *archived* record counts too. It is one unarchive away from
+/// being live beside this one, so treating the id as free here would create
+/// the very duplicate this guard promises the app never makes — and the user
+/// would meet it only later, when restoring the record they had put away.
 async fn ensure_new_record_is_unique(
   session : String,
   selected? : @pathx.Absolute,
@@ -88,6 +93,16 @@ async fn ensure_new_record_is_unique(
         "this conversation already has a record in \{root}; start a new one here",
       )
     }
+    if @fsx.is_dir(
+        @session_store.archived_root(root)
+        .join("sessions")
+        .join(session)
+        .to_path(),
+      ) {
+      raise EngineError(
+        "this conversation has an archived record in \{root}; start a new one here",
+      )
+    }
   }
 }
 

desktop/internal/engine/ops.mbt

@@ -250,6 +250,10 @@ pub async fn compact_run(
   // reused engine never builds one, and the store must be known either way.
   let store = requested_store(payload.workspace)
   @session_store.ensure_not_archived(session, Some(store))
+  // A compaction with no live engine spawns one, and a spawn writes the
+  // record: without this the command would open a second record for an id
+  // another store already owns, which `start_run` refuses.
+  ensure_new_record_is_unique(session, selected=store)
   let sessions = manager.serving()
   let slot = slot_for(sessions, session)
   // Claim before anything else — even with a live engine. A start between
@@ -352,6 +356,9 @@ pub async fn goal_run(
   // the slot is read — see `compact_run`.
   let store = requested_store(payload.workspace)
   @session_store.ensure_not_archived(session, Some(store))
+  // A goal is reachable before the first prompt, and its spawn writes the
+  // record just as a start's does — see `compact_run`.
+  ensure_new_record_is_unique(session, selected=store)
   let sessions = manager.serving()
   let slot = slot_for(sessions, session)
   // Claim before looking at the engine, exactly as compact_run does: this op

desktop/internal/engine/store_identity_wbtest.mbt

@@ -139,6 +139,89 @@ async test "a start cannot open a second record for an id another store owns" {
   @fs.rmdir(root.to_string(), recursive=true)
 }
 
+///|
+#cfg(not(platform="windows"))
+async test "no command opens a second record for an id another store owns" {
+  ambient_env_test_lock.acquire()
+  defer ambient_env_test_lock.release()
+  let previous = @sys.get_env_var("OPENSEEK_SESSION_ROOT")
+  let (root, actor) = store_identity_host("openseek-unique-commands-")
+  defer restore_session_root_env(previous)
+  let manager = actor.manager()
+  let owner = root.join("owner")
+  let ws = root.join("ws")
+  @fsx.ensure_dir(ws.to_path())
+  @fsx.ensure_dir(owner.to_path())
+  let sink = EventSink(fn(_) {  })
+  @async.with_task_group(group => {
+    group.spawn_bg(no_wait=true, allow_failure=true, () => {
+      actor.run(sink, fn(_) {  })
+    })
+    while manager.pump is Stopped {
+      @async.sleep(1)
+    }
+    ignore(@workspaces.add(owner.to_string(), fn(_) {  }))
+    ignore(@workspaces.add(ws.to_string(), fn(_) {  }))
+    // The first project owns `s-live`. A goal is reachable before the first
+    // prompt, and its spawn writes a record, so it must refuse for the same
+    // reason a start does rather than filing the second twin itself.
+    @fsx.ensure_dir(
+      @workspaces.store_root(owner).join("sessions/s-live").to_path(),
+    )
+    let goal_refusal = try {
+      ignore(
+        goal_run(manager, {
+          session: "s-live",
+          workspace: ws.to_string(),
+          text: Some("ship it"),
+          auto: None,
+          model: None,
+          max_steps: None,
+        }),
+      )
+      "no error"
+    } catch {
+      EngineError(detail) => detail
+      error if @async.is_being_cancelled() => raise error
+      error => "\{error}"
+    }
+    assert_true(goal_refusal.has_suffix("; start a new one here"))
+    guard manager.pump is Serving(sessions~) else { fail("pump stopped") }
+    assert_true(sessions.get("s-live") is None)
+    // An *archived* record in the other store counts as well: it is one
+    // unarchive away from standing beside whatever this start would create.
+    @fsx.ensure_dir(
+      @workspaces.store_root(owner)
+      .join("archived/sessions/s-put-away")
+      .to_path(),
+    )
+    let start_refusal = try {
+      ignore(
+        start_run(sink, manager, {
+          task: "fork me",
+          submission_id: None,
+          model: None,
+          max_steps: None,
+          session: "s-put-away",
+          workspace: ws.to_string(),
+        }),
+      )
+      "no error"
+    } catch {
+      EngineError(detail) => detail
+      error if @async.is_being_cancelled() => raise error
+      error => "\{error}"
+    }
+    assert_true(start_refusal.contains("archived record"))
+    assert_true(
+      start_refusal.contains(@workspaces.store_root(owner).to_string()),
+    )
+    assert_true(sessions.get("s-put-away") is None)
+    group.return_immediately(())
+  })
+  @fs.rmdir(root.to_string(), recursive=true)
+}
+
 ///|
 #cfg(not(platform="windows"))
 async test "an archived record in another store does not block a live one" {