fix(manager): recover non-SSH steward turns on concurrent registry sync and skip no-op global registry writes

loopx/chat_manager_context.py

@@ -799,6 +799,14 @@ def scope_unchanged() -> bool:
             )
         )
 
+    from .capabilities.manager_context.ssh_evidence import grants
+
+    inline_remote = bool(
+        remote_evidence
+        and grants(runtime_root, str(session.get("channel_id") or ""))
+    )
+    bound_steward = conversation_scope(session).get("bound_steward") is True
+
     for attempt in range(2):
         context = manager_turn_context(
             registry_path,
@@ -824,11 +832,30 @@ def scope_unchanged() -> bool:
             else None
         )
         if before != after or before_scope_id != after_scope_id:
+            if (
+                attempt == 0
+                and not inline_remote
+                and bound_steward
+                and registry_path is not None
+                and before is not None
+                and after is not None
+                and after_scope_id is not None
+                and set(before).issubset(set(after))
+                and manager_authorization_scope_id_for_registry(
+                    registry_path,
+                    before,
+                    runtime_root=runtime_root,
+                    channel_id=session.get("channel_id"),
+                )
+                == before_scope_id
+            ):
+                before, before_scope_id = after, after_scope_id
+                continue
             return unavailable_manager_context("external_authorization_changed")
         # Recollect local, on-demand context once under the original scope.
         # Inline remote collection owns one dial/Turn budget, including failed
         # reads; recollecting it would restart that budget.
-        if (attempt == 0 and not remote_evidence and before_scope_id is not None
+        if (attempt == 0 and not inline_remote and before_scope_id is not None
                 and context.get("warnings") == ["external_authorization_changed"]):
             continue
         break

loopx/global_registry.py

@@ -310,6 +310,12 @@ def merge_goal_entries(
                     raise ValueError(collision_message(collision))
                 action = "replaced-route"
             goal = preserve_attention_override(existing_goal, goal)
+            if (
+                "synced_at" in existing_goal
+                and {k: v for k, v in existing_goal.items() if k != "synced_at"}
+                == {k: v for k, v in goal.items() if k != "synced_at"}
+            ):
+                goal["synced_at"] = existing_goal["synced_at"]
         merged.append(goal)
         actions.append(f"{goal_id}:{action}")
         synced_ids.append(goal_id)
@@ -393,6 +399,12 @@ def _merge_global_registry_payload(
     if merged_projects:
         payload["projects"] = merged_projects
     payload["goals"] = merged_goals
+    if (
+        "updated_at" in existing
+        and {k: v for k, v in payload.items() if k != "updated_at"}
+        == {k: v for k, v in existing.items() if k != "updated_at"}
+    ):
+        payload["updated_at"] = existing["updated_at"]
     return payload, merged_goals, actions, synced_ids, collisions
 
 

tests/test_chat_manager_context.py

@@ -998,7 +998,7 @@ def test_trusted_manager_profile_rejects_endpoint_that_cannot_enforce_it(
 
     with pytest.raises(
         CodexChatAgentError,
-        match="requires the Codex endpoint",
+        match="requires the Codex or Antigravity endpoint",
     ) as caught:
         runtime._start_adapter(
             agent_id="claude-code",
@@ -1009,7 +1009,7 @@ def test_trusted_manager_profile_rejects_endpoint_that_cannot_enforce_it(
         )
 
     assert caught.value.error_code == "manager_runtime_endpoint_unsupported"
-    assert "Select the Codex Agent" in caught.value.gate["next_action"]
+    assert "Select the Codex or Antigravity Agent" in caught.value.gate["next_action"]
 
 
 @pytest.mark.parametrize("initial_profile,next_profile", [("restricted", "trusted_owner"), ("trusted_owner", "restricted")])
@@ -1652,3 +1652,64 @@ def read(goal, **kwargs):
     )
     assert revoked["goals"] == []
     assert reads == ["currently-authorized"]
+
+
+@pytest.mark.parametrize("mode", ["metadata_touch", "bound_steward_goal_added"])
+def test_remote_evidence_without_ssh_grants_recollects_on_concurrent_registry_update(
+    monkeypatch, tmp_path, mode
+):
+    """Non-Codex stewards pass remote_evidence=True; without SSH grants they still recover once."""
+    registry = tmp_path / "registry.json"
+    payload = {"goals": [{"id": "authorized", "goal_instance_id": "ginst_" + "a" * 32}]}
+    registry.write_text(json.dumps(payload))
+    collect = context.build_goal_portfolio
+    calls = []
+    session = {
+        "goal_id": "loopx-manager",
+        "channel_id": "manager.external.native." + "b" * 24 + "." + "c" * 24,
+        "steward_context": {
+            "kind": "bound_steward",
+            "project_ref": "a" * 24,
+            "workspace_path": str(tmp_path),
+            "audience": "bound_owner",
+            "grant": "portfolio_read",
+            "binding_id": "b" * 24,
+            "source_ref": "c" * 24,
+            "provider_ref": "d" * 24,
+            "operator_ref": "e" * 24,
+            "goal_ids": ["authorized"],
+        },
+    }
+
+    def racing_collection(**kwargs):
+        result = collect(**kwargs)
+        calls.append(result["inventory_revision"])
+        if len(calls) == 1:
+            if mode == "metadata_touch":
+                payload["updated_at"] = "2026-11-02T07:41:53Z"
+            else:
+                payload["goals"].append(
+                    {"id": "newly-registered", "goal_instance_id": "ginst_" + "b" * 32}
+                )
+            registry.write_text(json.dumps(payload))
+        return result
+
+    monkeypatch.setattr(context, "build_goal_portfolio", racing_collection)
+    result = context.collect_manager_turn_context(
+        registry,
+        session,
+        tmp_path,
+        lambda _s: [g["id"] for g in json.loads(registry.read_text())["goals"]],
+        include_details=False,
+        remote_evidence=True,
+    )
+    assert len(calls) == 2
+    assert result.get("warnings") != ["external_authorization_changed"]
+    expected_ids = (
+        ["authorized"]
+        if mode == "metadata_touch"
+        else ["authorized", "newly-registered"]
+    )
+    assert [g["goal_id"] for g in result["goals"]] == expected_ids
+    assert result.get("authorization_scope_id")
+

tests/test_global_registry_write_serialization.py

@@ -1170,3 +1170,55 @@ def test_concurrent_project_syncs_do_not_drop_goals(tmp_path: Path) -> None:
     assert {
         str(goal.get("id")): str(goal.get("project_id")) for goal in payload["goals"]
     } == {f"goal-{name}": f"project-{name}" for name in names}
+
+
+def test_redundant_sync_does_not_rewrite_unchanged_global_registry(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Routine refresh-state syncs must not churn global registry timestamps when nothing changed."""
+    runtime_root = tmp_path / "runtime"
+    registry_path = _project_registry(tmp_path, "alpha", runtime_root)
+    global_path = global_registry_path(runtime_root)
+
+    ticks = iter(["2026-11-02T07:41:00+00:00", "2026-11-02T07:41:53+00:00", "2026-11-02T07:42:10+00:00"])
+    monkeypatch.setattr(global_registry, "now_local", lambda: next(ticks))
+
+    first = sync_project_registry_to_global(
+        registry_path=registry_path,
+        runtime_root_override=str(runtime_root),
+        dry_run=False,
+    )
+    assert first["ok"] is True
+    before_bytes = global_path.read_bytes()
+
+    writes: list[Path] = []
+    real_write = global_registry.write_json
+
+    def tracking_write(path: Path, payload: dict[str, Any]) -> None:
+        writes.append(path)
+        real_write(path, payload)
+
+    monkeypatch.setattr(global_registry, "write_json", tracking_write)
+
+    second = sync_project_registry_to_global(
+        registry_path=registry_path,
+        runtime_root_override=str(runtime_root),
+        dry_run=False,
+    )
+    assert second["ok"] is True
+    assert writes == []
+    assert global_path.read_bytes() == before_bytes
+
+    source_payload = json.loads(registry_path.read_text(encoding="utf-8"))
+    source_payload["goals"][0]["objective"] = "updated objective"
+    registry_path.write_text(json.dumps(source_payload), encoding="utf-8")
+
+    third = sync_project_registry_to_global(
+        registry_path=registry_path,
+        runtime_root_override=str(runtime_root),
+        dry_run=False,
+    )
+    assert third["ok"] is True
+    assert writes == [global_path]
+    assert global_path.read_bytes() != before_bytes
+