diff --git a/docs/assets/personal-workspace/steward-reports/allocated-fallback-desktop.png b/docs/assets/personal-workspace/steward-reports/allocated-fallback-desktop.png new file mode 100644 index 0000000000..bcf79dbf8b Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/allocated-fallback-desktop.png differ diff --git a/docs/assets/personal-workspace/steward-reports/allocated-fallback-narrow.png b/docs/assets/personal-workspace/steward-reports/allocated-fallback-narrow.png new file mode 100644 index 0000000000..9a8459508a Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/allocated-fallback-narrow.png differ diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index bb27515b3b..1ac75f66d3 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -29,6 +29,29 @@ readback and executor compatibility, not scheduled generation or live delivery. ![The same retained schedule at 390 pixels](../assets/personal-workspace/steward-reports/retained-executor-narrow.png) +A preferred executor is distinct from the executor already allocated to the +original Session. When Core has retained a private Antigravity steward on its +qualified, flexible Codex fallback, the existing monitor and report controls +remain usable. Bound input, recovery and native consent share the same executor +match; the settings API observes that Core decision instead of rejecting the +configured preference. This neither switches hosts nor enables native scheduled +work on Antigravity. A preference alone, an unallocated historical thread or a +pinned policy does not qualify the fallback. Original source, grant, actor, +idle-queue and report-window checks still apply on every operation. + +The same disposable journey accepts `qualify-codex-fallback` to seed this +retained allocation. With the same packaged UI, the baseline backend disables +all four controls; the current backend preserves the actor and window and +enables them. An unavailable synthetic original source rejects a pause without +changing the schedule. Restoring it permits pause and re-enable through the +existing UI. These views qualify configuration and admission, not model quality +or an installed scheduled delivery; the fixture's unchanged endpoint-pool +configuration still refuses a mismatched model dispatch. + +![Qualified retained fallback in desktop settings](../assets/personal-workspace/steward-reports/allocated-fallback-desktop.png) + +![The same qualified fallback at 390 pixels](../assets/personal-workspace/steward-reports/allocated-fallback-narrow.png) + Markdown, the HTML report and its copied full text use the same source-link rule. An unchanged, well-formed HTTP(S) address is a link. Opaque evidence digests, native Task/artifact references and private relative paths stay source diff --git a/examples/personal-workspace-browser/steward_report_fixture.py b/examples/personal-workspace-browser/steward_report_fixture.py index 2bb42cc1c3..832010df4f 100644 --- a/examples/personal-workspace-browser/steward_report_fixture.py +++ b/examples/personal-workspace-browser/steward_report_fixture.py @@ -174,9 +174,14 @@ def fail_report(_adapter, _prompt, event_sink): restart_tokens = set() restart_refusals = set() executor_restored = False + executor_allocated = False try: while not stop.wait(.2): server.assets_dir = current_assets if (output / "use-current-assets").exists() else baseline_assets + if args.retained_executor and not executor_allocated and (output / "qualify-codex-fallback").exists(): + from test_native_steward_managed_mode import retain_codex_fallback + retain_codex_fallback(store, runtime, sid) + executor_allocated = True if args.retained_executor and not executor_restored and (output / "restore-supported-executor").exists(): current = transport.bindings.read() selected = next(row for row in current["bindings"] if row["binding_id"] == binding["binding_id"]) @@ -263,6 +268,7 @@ def fail_report(_adapter, _prompt, event_sink): "existing_wake_owner_running": wake is not None, "existing_wake_owner_starts": owner_start_count, "retained_executor_restored": executor_restored, + "retained_executor_allocated": executor_allocated, "runtime_restarts": len(restart_tokens), "runtime_restart_rejections": len(restart_refusals), **({"managed_source_present": artifact.exists(), diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index dbac78cd2e..5d8209cf19 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -162,6 +162,9 @@ def health(self, binding_id: str) -> dict[str, Any]: cadence = monitor_cadence_delta(monitor.get("cadence")) if monitor else None from .steward_reports import StewardReports return {"available": True, "prepared": True, + **(self.bindings._core("collaboration.steward.managed_mode.executor", { + "binding": binding, "session": session}) + if binding["executor_endpoint_id"] != "codex" else {}), "scheduling_enabled": bool(active and monitor and monitor.get("status") == "open"), "goal_active": active, "report": StewardReports(self.controller).health(binding_id), "monitor": ({"todo_id": monitor["todo_id"], "status": monitor["status"], diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index f21e5bd9cd..2df7baba14 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -5,7 +5,7 @@ import {BARE_SHA256_PATTERN} from "../content_digest.ts"; import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; import {requireInteger, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; import {requireAutomationCadenceEligibility} from "../quota/automation_cadence.ts"; -import {normalizeProjectContext, normalizeStewardContext, normalizeStewardGoalScope} from "./conversation_scope.ts"; +import {boundConversationExecutorMatches, normalizeProjectContext, normalizeStewardContext, normalizeStewardGoalScope} from "./conversation_scope.ts"; import {bindStewardManagedActor, normalizeStewardManagedActor} from "./steward_managed_mode.ts"; /** Core owns the context and audience grant. A provider supplies verified, @@ -323,21 +323,6 @@ function boundProject(params: JsonObject, row: JsonObject): Record 0 - && Array.isArray(allocation.eligible_endpoints) && allocation.eligible_endpoints.length === 2 - && allocation.eligible_endpoints.includes("antigravity") && allocation.eligible_endpoints.includes("codex") - && Array.isArray(session.upstream_context_history) && session.upstream_context_history.some(value => - value && typeof value === "object" && !Array.isArray(value) && (value as JsonObject).agent_id === "antigravity" - && typeof (value as JsonObject).thread_id === "string"); -} export function resolveBoundConversation(params: JsonObject): JsonObject { const current = state(params.current); @@ -394,7 +379,7 @@ export function resolveBoundConversation(params: JsonObject): JsonObject { sessionIdentity(session.session_id, "bound Session identity"); if (session.status === "closed" || session.channel_id !== channel || session.goal_id !== (steward ? "loopx-manager" : null) - || !boundExecutorMatches(row, session) + || !boundConversationExecutorMatches(row, session) || session.session_mode !== "managed_runtime" || (steward ? session.project_context : session.steward_context) != null || params.session_context === undefined || !isDeepStrictEqual(steward ? session.steward_context : session.project_context, params.session_context)) { @@ -634,7 +619,7 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO const selected = binding(params.binding); if (selected.audience === "group") return no; const steward = selected.context_kind === "steward"; - if (!boundExecutorMatches(selected, {...current, executor_endpoint_id: current.executor_endpoint_id ?? current.agent_id}) || current.goal_id !== (steward ? "loopx-manager" : null)) return no; + if (!boundConversationExecutorMatches(selected, {...current, executor_endpoint_id: current.executor_endpoint_id ?? current.agent_id}) || current.goal_id !== (steward ? "loopx-manager" : null)) return no; const saved = steward ? normalizeStewardContext(current.steward_context) : normalizeProjectContext(current.project_context); const source = requireJsonObject(row.source, "original conversation source"); diff --git a/loopx/control_plane/collaboration/conversation_scope.ts b/loopx/control_plane/collaboration/conversation_scope.ts index e878572da4..c29e048e1c 100644 --- a/loopx/control_plane/collaboration/conversation_scope.ts +++ b/loopx/control_plane/collaboration/conversation_scope.ts @@ -215,3 +215,20 @@ export function resolveConversationScope(input: Record): Conver } return {kind: "unavailable", goal_ids: [], private_conversation: false}; } + +/** One executor match for bound input and native work; provider/source admission remains separate. */ +export function boundConversationExecutorMatches(row: Record, session: Record): boolean { + if (session.agent_id === row.executor_endpoint_id && session.executor_endpoint_id === row.executor_endpoint_id) return true; + const allocation = session.manager_executor_allocation as Record | undefined; + return row.context_kind === "steward" && row.audience !== "group" + && row.executor_endpoint_id === "antigravity" && session.agent_id === "codex" + && session.executor_endpoint_id === "codex" && allocation?.schema_version === "manager_executor_allocation_v0" + && allocation.selection_policy === "flexible" && allocation.configured_endpoint === "antigravity" + && allocation.executor_endpoint === "codex" && allocation.allocation_reason === "flexible_availability_fallback" + && typeof allocation.configuration_revision === "string" && allocation.configuration_revision.length > 0 + && Array.isArray(allocation.eligible_endpoints) && allocation.eligible_endpoints.length === 2 + && allocation.eligible_endpoints.includes("antigravity") && allocation.eligible_endpoints.includes("codex") + && Array.isArray(session.upstream_context_history) && session.upstream_context_history.some(value => + value && typeof value === "object" && !Array.isArray(value) && (value as Record).agent_id === "antigravity" + && typeof (value as Record).thread_id === "string"); +} diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index e64523a058..852500b675 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -9,7 +9,7 @@ import type { JsonObject } from "../runtime_decode.ts"; import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; -import {goalAgentConversationIdentity, resolveConversationScope} from "./conversation_scope.ts"; +import {boundConversationExecutorMatches, goalAgentConversationIdentity, resolveConversationScope} from "./conversation_scope.ts"; import {readTurnSelectionRejection} from "../turn_driver/selection_rejection.ts"; import {EffectiveAction} from "../quota/effective_action.generated.ts"; @@ -415,6 +415,15 @@ export function normalizeStewardManagedActor(value: unknown): JsonObject { creation_operation_id: text(row.creation_operation_id, "managed Goal creation")}; } +/** Supported native host metadata, never authorization or a request to switch hosts. */ +export function observeStewardNativeExecutor(params: JsonObject): JsonObject { + const binding = requireJsonObject(params.binding, "steward binding"); + const session = requireJsonObject(params.session, "original steward Session"); + const supported = session.agent_id === "codex" && (binding.executor_endpoint_id === "codex" + || boundConversationExecutorMatches(binding, session)); + return {execution_available: supported, ...(!supported ? {execution_error_code: "unsupported_executor"} : {})}; +} + export function stewardManagedBasis(params: JsonObject, executing = false) { const binding = requireJsonObject(params.binding, "steward binding"); const session = requireJsonObject(params.session, "original steward Session"); @@ -423,8 +432,8 @@ export function stewardManagedBasis(params: JsonObject, executing = false) { const id = text(binding.binding_id, "binding identity", /^[a-f0-9]{24}$/); const source = text(context.source_ref, "steward source", /^[a-f0-9]{24}$/); if (binding.context_kind !== "steward" || binding.grant !== "portfolio_read" || binding.enabled !== true - || binding.audience === "group" || binding.executor_endpoint_id !== "codex" - || session.goal_id !== "loopx-manager" || session.agent_id !== "codex" + || binding.audience === "group" || observeStewardNativeExecutor(params).execution_available !== true + || session.goal_id !== "loopx-manager" || session.session_mode === "attached_host" || session.status === "closed" || session.channel_id !== `manager.external.native.${id}.${source}` || !executing && (session.active_turn_id || params.queued_turn_count !== 0)) { diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index ed580b4f20..079bc62bb0 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -594,6 +594,7 @@ export function createEffectRuntimeHandlers( ["collaboration.steward.command", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({stewardCommand}) => stewardCommand)], ["collaboration.steward.authorize_creation", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({authorizeStewardCreation}) => authorizeStewardCreation)], ["collaboration.steward.managed_mode.plan", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({planStewardManagedMode}) => planStewardManagedMode)], + ["collaboration.steward.managed_mode.executor", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({observeStewardNativeExecutor}) => observeStewardNativeExecutor)], ["collaboration.steward.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitStewardManagedTurn}) => admitStewardManagedTurn)], ["collaboration.session.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitManagedSessionTurn}) => admitManagedSessionTurn)], ["collaboration.session.managed_identity", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitManagedSessionIdentity}) => admitManagedSessionIdentity)], diff --git a/loopx/extensions/lark/private_conversation_api.py b/loopx/extensions/lark/private_conversation_api.py index 293b01b515..efecf1c545 100644 --- a/loopx/extensions/lark/private_conversation_api.py +++ b/loopx/extensions/lark/private_conversation_api.py @@ -58,10 +58,7 @@ def _private_conversations(self) -> None: and hasattr(self.server.lark_private_conversations, "ingress") else {}), **({"owner_attention": self.server.lark_private_conversations.owner_attention.health(row["binding_id"])} if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" else {}), - **({"managed_steward": { - **self.server.lark_private_conversations.managed_steward.health(row["binding_id"]), - **({"execution_available": False, "execution_error_code": "unsupported_executor"} - if row["executor_endpoint_id"] != "codex" else {})}} + **({"managed_steward": self.server.lark_private_conversations.managed_steward.health(row["binding_id"])} if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" and (row["executor_endpoint_id"] == "codex" or row.get("managed_actor")) and hasattr(self.server.lark_private_conversations, "managed_steward") else {}), diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index a758466e86..c938d3314e 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -127,6 +127,69 @@ function original() { return {...params, binding: {...binding, managed_actor: actor}, actor, goal_active: true, source}; } +function allocatedFallback() { + const p = original(); + const binding = {...p.binding, executor_endpoint_id: "antigravity"}; + const session = {...p.session, executor_endpoint_id: "codex", + steward_context: {...p.session.steward_context, executor_endpoint_id: "antigravity"}, + manager_executor_allocation: {schema_version: "manager_executor_allocation_v0", selection_policy: "flexible", + configured_endpoint: "antigravity", executor_endpoint: "codex", configuration_revision: "original-profile", + allocation_reason: "flexible_availability_fallback", eligible_endpoints: ["antigravity", "codex"]}, + upstream_context_history: [{agent_id: "antigravity", thread_id: "original-native-thread", + manager_context_version: null, agent_model_id: "gemini-4-argon-high"}]}; + return {...p, binding, session}; +} + +test("a Core-qualified Codex fallback retains report consent and native report identity", () => { + const p = allocatedFallback(); + const route = {binding_id: p.binding.binding_id, source_ref: p.source, session_id: p.session.session_id, + provider_ref: p.binding.provider_ref, operator_ref: p.binding.operator_ref}; + const proposed = {enabled: true, profile_preset: "portfolio-progress", + route_ref: `steward-private:${p.binding.binding_id}:${p.source}`}; + const input = {...p, configuration: null, proposed, route, operation: "enable"}; + const before = structuredClone(input); + assert.deepEqual(planStewardReport(input).actor, p.actor); + assert.deepEqual(input, before); + const window = {window_id: "cadence_original", subscription_revision: "sha256:original", + goal_id: p.actor.goal_id, agent_id: p.actor.agent_id}; + const request = {operation: "native_turn", host: "codex-app-server", actor: p.actor, + turn_instance_id: `report.${window.window_id}`, todo_id: "todo_report", validation_command: ["validate"], + work_kind: "report", report_window_id: window.window_id, report_subscription_revision: window.subscription_revision}; + const params = {...p, request, report_window: window, + report_subscription: {...proposed, effective_revision: window.subscription_revision}, + report_todo: {todo_id: request.todo_id, target_key: window.window_id, action_kind: "generate_periodic_report", + claimed_by: "steward", status: "open"}}; + assert.deepEqual(admitStewardManagedTurn(params).actor, p.actor); + for (const changed of [{binding: {...p.binding, enabled: false}}, {binding: {...p.binding, audience: "group"}}, + {session: {...p.session, status: "closed"}}, {queued_turn_count: 1}, {goal_active: false}, + {report_subscription: {...params.report_subscription, enabled: false}}, + {report_window: {...window, window_id: "another-window"}}, + {observation: {...p.observation, operator_ref: "0".repeat(24)}}]) { + assert.throws(() => admitStewardManagedTurn({...params, ...changed})); + } +}); + +test("a configured preference alone cannot qualify a different native report executor", () => { + const p = allocatedFallback(); + const proposed = {enabled: true, profile_preset: "portfolio-progress", + route_ref: `steward-private:${p.binding.binding_id}:${p.source}`}; + const route = {binding_id: p.binding.binding_id, source_ref: p.source, session_id: p.session.session_id, + provider_ref: p.binding.provider_ref, operator_ref: p.binding.operator_ref}; + const input = {...p, configuration: null, proposed, route, operation: "enable"}; + const allocation = p.session.manager_executor_allocation; + for (const changed of [{selection_policy: "pinned"}, {configured_endpoint: "codex"}, + {executor_endpoint: "antigravity"}, {allocation_reason: "configured_preference"}, + {configuration_revision: ""}, {eligible_endpoints: ["codex"]}, + {eligible_endpoints: ["codex", "antigravity", "other"]}]) { + assert.throws(() => planStewardReport({...input, + session: {...p.session, manager_executor_allocation: {...allocation, ...changed}}})); + } + for (const changed of [{upstream_context_history: []}, {executor_endpoint_id: "antigravity"}, + {agent_id: "antigravity"}, {session_mode: "attached_host"}]) { + assert.throws(() => planStewardReport({...input, session: {...p.session, ...changed}})); + } +}); + test("report consent fixes the original actor, source and private audience without starting work", () => { const p = original(); const proposed = {enabled: true, profile_preset: "portfolio-progress", diff --git a/tests/test_native_steward_managed_mode.py b/tests/test_native_steward_managed_mode.py index 0743c38f1e..2ba68803c4 100644 --- a/tests/test_native_steward_managed_mode.py +++ b/tests/test_native_steward_managed_mode.py @@ -25,6 +25,23 @@ def original(steward): return row["session_id"], transport.managed_steward +def retain_codex_fallback(store, runtime, sid): + """Seed trusted allocation metadata only in disposable native/HTTP fixtures.""" + from loopx.chat_manager import manager_executor_allocation + + session = store.load_session(sid) + assert session["agent_id"] == session["executor_endpoint_id"] == "codex" + allocation = session.get("manager_executor_allocation") or manager_executor_allocation(runtime, environ={}) + store.update_session(sid, manager_executor_allocation={**allocation, + "selection_policy": "flexible", "configured_endpoint": "antigravity", + "executor_endpoint": "codex", "eligible_endpoints": ["antigravity", "codex"], + "allocation_reason": "flexible_availability_fallback"}, + upstream_context_history=[*(session.get("upstream_context_history") or []), + {"agent_id": "antigravity", "thread_id": "original-native-thread", + "manager_context_version": None, "agent_model_id": "gemini-4-argon-high"}]) + return allocation + + def storage(runtime, provider): args = {"runtime_root": runtime.coordination_runtime_root, "registry": build_builtin_machine_configuration_registry(), @@ -287,6 +304,20 @@ def request(path, body=None, origin=None): selected = next(row for row in current["bindings"] if row["binding_id"] == binding["binding_id"]) selected["executor_endpoint_id"] = "antigravity" transport.bindings.path.write_text(json.dumps(current)) + # Core may already have retained this same Session on its qualified + # Codex fallback. Neither the UI nor native consent chooses a new host. + allocation = retain_codex_fallback(store, runtime, sid) + fallback = next(row for row in request(base)[1]["connections"] + if row["binding_id"] == binding["binding_id"])["managed_steward"] + assert fallback.pop("execution_available") is True + assert "execution_error_code" not in fallback and fallback == expected + assert request(route, {"binding_id": binding["binding_id"]})[0] == 200 + assert request(report_route, report_setting)[0] == 200 + assert request(monitor_route, {**setting, "operation": "enable"})[0] == 200 + assert journal_path.read_bytes() == journal + assert store.messages(sid) == before_history and capture.read_bytes() == before_calls + # A historical thread alone cannot qualify an unallocated executor. + store.update_session(sid, manager_executor_allocation=allocation) original_binding, registry = transport.bindings.path.read_bytes(), runtime.registry_path.read_bytes() row = next(row for row in request(base)[1]["connections"] if row["binding_id"] == binding["binding_id"]) state = row["managed_steward"]