From 85bd51805a2e7c96f24ba12a069128480cd9ab80 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Wed, 23 Sep 2026 19:44:06 +0800 Subject: [PATCH 1/3] fix(goals): bind lifecycle confirmation to source Signed-off-by: duanjialing.777 --- loopx/chat_actions.py | 6 + loopx/chat_goal_lifecycle_actions.py | 56 ++++++- tests/control_plane/test_goal_activation.py | 158 ++++++++++++++++++++ 3 files changed, 218 insertions(+), 2 deletions(-) diff --git a/loopx/chat_actions.py b/loopx/chat_actions.py index 24fbf24b88..5fd41f5fa8 100644 --- a/loopx/chat_actions.py +++ b/loopx/chat_actions.py @@ -1172,6 +1172,12 @@ def preview(self, request: Mapping[str, Any]) -> dict[str, Any]: ) evidence = ["The recoverable Goal and Agent Chat Session is available."] permission = "scoped_correction" + elif action_kind == "goal.lifecycle": + fingerprint = self._goal_lifecycle_preview_fingerprint(normalized) + evidence = [ + "The lifecycle transition is bound to the current authoritative Goal source." + ] + permission = "durable_write" elif action_kind == "team.plan": # A plan is reviewed against this Goal's registration facts *and* # the intent its lanes would advance, so both are bound here and diff --git a/loopx/chat_goal_lifecycle_actions.py b/loopx/chat_goal_lifecycle_actions.py index 8ce34aae72..b54a2ca577 100644 --- a/loopx/chat_goal_lifecycle_actions.py +++ b/loopx/chat_goal_lifecycle_actions.py @@ -12,6 +12,35 @@ class ChatGoalLifecycleActionMixin: """Keep Goal activation policy separate from general action orchestration.""" + def _goal_lifecycle_preview_fingerprint( + self, + parameters: dict[str, Any], + ) -> str: + operation = str(parameters["operation"]) + if operation == "delete": + return self._registry_fingerprint() + target_state = ( + GoalActivationState.STOPPED + if operation == "stop" + else GoalActivationState.ACTIVE + ) + preview = set_goal_activation_state( + registry_path=self.registry_path, + goal_id=str(parameters["goal_id"]), + state=target_state, + reason=parameters.get("reason"), + execute=False, + ) + fingerprint = str(preview.get("observed_state_fingerprint") or "") + if not preview.get("ok") or not fingerprint: + raise ValueError( + str( + preview.get("error") + or "Goal lifecycle source fingerprint is unavailable" + ) + ) + return fingerprint + def _normalize_goal_lifecycle( self, parameters: dict[str, Any] ) -> dict[str, Any]: @@ -94,10 +123,10 @@ def _apply_goal_lifecycle( ) -> dict[str, Any]: from .chat_actions import _digest - current_fingerprint = self._registry_fingerprint() goal_id = str(parameters["goal_id"]) operation = str(parameters["operation"]) if operation == "delete": + current_fingerprint = self._registry_fingerprint() return self._apply_goal_delete( proposal_id, proposal, goal_id, current_fingerprint ) @@ -107,8 +136,16 @@ def _apply_goal_lifecycle( if operation == "stop" else GoalActivationState.ACTIVE ) - current_state = goal_activation_state(self._goal(goal_id)) expected_fingerprint = str(proposal.get("expected_state_fingerprint") or "") + current = set_goal_activation_state( + registry_path=self.registry_path, + goal_id=goal_id, + state=target_state, + reason=parameters.get("reason"), + execute=False, + ) + current_fingerprint = str(current.get("observed_state_fingerprint") or "") + current_state = GoalActivationState(str(current.get("before_state") or "")) idempotent_reapply = ( current_state is target_state and current_fingerprint != expected_fingerprint @@ -126,8 +163,23 @@ def _apply_goal_lifecycle( state=target_state, reason=parameters.get("reason"), actor_kind="owner", + expected_state_fingerprint=( + current_fingerprint + if idempotent_reapply + else expected_fingerprint + ), execute=True, ) + if result.get("error_kind") == "goal_action_stale": + stale = self.store.apply( + proposal_id, + current_state_fingerprint=str( + result.get("observed_state_fingerprint") + or current_fingerprint + ), + receipt={}, + ) + return {"proposal": stale, "turn": None} if not result.get("ok") or not (result.get("readback") or {}).get( "verified" ): diff --git a/tests/control_plane/test_goal_activation.py b/tests/control_plane/test_goal_activation.py index e07bbe4b90..c54ac09a72 100644 --- a/tests/control_plane/test_goal_activation.py +++ b/tests/control_plane/test_goal_activation.py @@ -5,6 +5,7 @@ import pytest +from loopx import chat_goal_lifecycle_actions from loopx.chat_action_store import ChatActionStore from loopx.chat_actions import ChatActionService from loopx.control_plane.goals.activation import ( @@ -416,6 +417,163 @@ def test_owner_confirmed_typed_action_stops_goal( assert goal_activation_state(_goal(global_registry)) is GoalActivationState.STOPPED +@pytest.mark.parametrize( + ("operation", "initial_state"), + [ + ("stop", GoalActivationState.ACTIVE), + ("resume", GoalActivationState.STOPPED), + ], +) +def test_owner_confirmed_lifecycle_action_rejects_a_changed_source_registry( + connected_registries: tuple[Path, Path], + tmp_path: Path, + operation: str, + initial_state: GoalActivationState, +) -> None: + source_registry, global_registry = connected_registries + if initial_state is GoalActivationState.STOPPED: + stopped = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state=GoalActivationState.STOPPED, + actor_kind="owner", + execute=True, + ) + assert stopped["ok"] is True + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": f"{operation.title()} a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": operation, + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": f"{operation}-goal-one-before-source-change", + } + ) + source_payload = load_registry(source_registry) + registry_goals(source_payload)[0]["display_name"] = "Changed after confirmation" + _write_json(source_registry, source_payload) + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert ( + goal_activation_state(_goal(source_registry)) + is initial_state + ) + assert ( + goal_activation_state(_goal(global_registry)) + is initial_state + ) + + +def test_owner_confirmed_lifecycle_action_rechecks_source_inside_write_lock( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_registry, global_registry = connected_registries + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Stop a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "stop", + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "stop-goal-one-before-lock-race", + } + ) + original = chat_goal_lifecycle_actions.set_goal_activation_state + source_changed = False + + def change_source_before_execute(**kwargs: object) -> dict[str, object]: + nonlocal source_changed + if kwargs.get("execute") is True and not source_changed: + source_payload = load_registry(source_registry) + registry_goals(source_payload)[0]["display_name"] = ( + "Changed before lock acquisition" + ) + _write_json(source_registry, source_payload) + source_changed = True + return original(**kwargs) + + monkeypatch.setattr( + chat_goal_lifecycle_actions, + "set_goal_activation_state", + change_source_before_execute, + ) + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert ( + goal_activation_state(_goal(source_registry)) + is GoalActivationState.ACTIVE + ) + assert ( + goal_activation_state(_goal(global_registry)) + is GoalActivationState.ACTIVE + ) + + +def test_owner_confirmed_lifecycle_action_recovers_after_committed_stop( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + _source_registry, global_registry = connected_registries + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Stop a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "stop", + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "stop-goal-one-response-loss", + } + ) + committed = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state=GoalActivationState.STOPPED, + reason="Owner confirmed from the workspace", + actor_kind="owner", + execute=True, + ) + assert committed["ok"] is True + + recovered = service.apply(str(proposal["proposal_id"])) + + assert recovered["proposal"]["status"] == "applied" + assert recovered["proposal"]["receipt"]["outcome"] == "goal_already_stopped" + assert ( + goal_activation_state(_goal(global_registry)) + is GoalActivationState.STOPPED + ) + + def test_delete_stopped_goal_removes_source_and_global( connected_registries: tuple[Path, Path], ) -> None: From 138d6575c45913a898c3ce6afb1410d9955dec31 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Thu, 24 Sep 2026 14:01:30 +0800 Subject: [PATCH 2/3] fix(goals): bind lifecycle actions to source route Signed-off-by: duanjialing.777 --- loopx/chat_actions.py | 4 +- loopx/chat_goal_lifecycle_actions.py | 37 ++- loopx/cli_commands/goal_lifecycle.py | 2 +- .../control_plane/goals/activation_service.py | 212 +++++++++++++----- loopx/control_plane/goals/operator_actions.py | 8 +- loopx/global_registry.py | 76 ++++--- tests/control_plane/test_goal_activation.py | 204 +++++++++++++++++ 7 files changed, 453 insertions(+), 90 deletions(-) diff --git a/loopx/chat_actions.py b/loopx/chat_actions.py index 0dd86d0143..97cc005455 100644 --- a/loopx/chat_actions.py +++ b/loopx/chat_actions.py @@ -1163,7 +1163,9 @@ def preview(self, request: Mapping[str, Any]) -> dict[str, Any]: evidence = ["The recoverable Goal and Agent Chat Session is available."] permission = "scoped_correction" elif action_kind == "goal.lifecycle": - fingerprint = self._goal_lifecycle_preview_fingerprint(normalized) + lifecycle_preview = self._goal_lifecycle_preview(normalized) + fingerprint = str(lifecycle_preview["state_fingerprint"]) + canonical_update_basis = lifecycle_preview.get("source_basis") evidence = [ "The lifecycle transition is bound to the current authoritative Goal source." ] diff --git a/loopx/chat_goal_lifecycle_actions.py b/loopx/chat_goal_lifecycle_actions.py index b54a2ca577..2560ade547 100644 --- a/loopx/chat_goal_lifecycle_actions.py +++ b/loopx/chat_goal_lifecycle_actions.py @@ -9,16 +9,19 @@ from .control_plane.goals.deletion_service import delete_stopped_goal +GOAL_LIFECYCLE_SOURCE_BASIS_SCHEMA_VERSION = "loopx_goal_lifecycle_source_basis_v1" + + class ChatGoalLifecycleActionMixin: """Keep Goal activation policy separate from general action orchestration.""" - def _goal_lifecycle_preview_fingerprint( + def _goal_lifecycle_preview( self, parameters: dict[str, Any], - ) -> str: + ) -> dict[str, Any]: operation = str(parameters["operation"]) if operation == "delete": - return self._registry_fingerprint() + return {"state_fingerprint": self._registry_fingerprint()} target_state = ( GoalActivationState.STOPPED if operation == "stop" @@ -32,14 +35,21 @@ def _goal_lifecycle_preview_fingerprint( execute=False, ) fingerprint = str(preview.get("observed_state_fingerprint") or "") - if not preview.get("ok") or not fingerprint: + source_identity = str(preview.get("source_identity") or "") + if not preview.get("ok") or not fingerprint or not source_identity: raise ValueError( str( preview.get("error") - or "Goal lifecycle source fingerprint is unavailable" + or "Goal lifecycle source identity is unavailable" ) ) - return fingerprint + return { + "state_fingerprint": fingerprint, + "source_basis": { + "schema_version": GOAL_LIFECYCLE_SOURCE_BASIS_SCHEMA_VERSION, + "source_identity": source_identity, + }, + } def _normalize_goal_lifecycle( self, parameters: dict[str, Any] @@ -146,8 +156,21 @@ def _apply_goal_lifecycle( ) current_fingerprint = str(current.get("observed_state_fingerprint") or "") current_state = GoalActivationState(str(current.get("before_state") or "")) + source_basis = proposal.get("canonical_update_basis") + expected_source_identity = ( + str(source_basis.get("source_identity") or "") + if isinstance(source_basis, dict) + and source_basis.get("schema_version") + == GOAL_LIFECYCLE_SOURCE_BASIS_SCHEMA_VERSION + else "" + ) + source_route_matches = bool( + expected_source_identity + and expected_source_identity == current.get("source_identity") + ) idempotent_reapply = ( - current_state is target_state + source_route_matches + and current_state is target_state and current_fingerprint != expected_fingerprint ) if current_fingerprint != expected_fingerprint and not idempotent_reapply: diff --git a/loopx/cli_commands/goal_lifecycle.py b/loopx/cli_commands/goal_lifecycle.py index 784e41da16..22a74b83b3 100644 --- a/loopx/cli_commands/goal_lifecycle.py +++ b/loopx/cli_commands/goal_lifecycle.py @@ -45,7 +45,7 @@ def register_goal_lifecycle_command( ) parser.add_argument( "--expected-state-fingerprint", - help="SHA-256 registry fingerprint from a fresh goal-actions projection.", + help="Source-bound SHA-256 fingerprint from a fresh goal-actions projection.", ) parser.add_argument( "--execute", diff --git a/loopx/control_plane/goals/activation_service.py b/loopx/control_plane/goals/activation_service.py index cc4eda6da0..d0ece3f5d3 100644 --- a/loopx/control_plane/goals/activation_service.py +++ b/loopx/control_plane/goals/activation_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import ExitStack from dataclasses import dataclass from enum import Enum import hashlib @@ -8,6 +9,8 @@ from typing import Any from ..projects.registry_codec import project_registry_transaction +from ...configuration_transaction import configuration_payload_revision +from ...file_lock import exclusive_file_lock from ...global_registry import sync_project_registry_to_global from ...history import load_registry from ...registry import registry_goals @@ -25,6 +28,9 @@ GOAL_ACTIVATION_TRANSITION_SCHEMA_VERSION = "loopx_goal_activation_transition_v1" GOAL_ACTIVATION_READBACK_SCHEMA_VERSION = "loopx_goal_activation_readback_v1" +GOAL_ACTIVATION_SOURCE_FINGERPRINT_SCHEMA_VERSION = ( + "loopx_goal_activation_source_fingerprint_v1" +) GOAL_ACTIVATION_AUTHORITY_ROUTE_SCHEMA_VERSION = ( "loopx_goal_activation_authority_route_v1" ) @@ -71,6 +77,41 @@ def _same_path(left: Path, right: Path) -> bool: return str(left.expanduser()) == str(right.expanduser()) +def _goal_activation_source_identity(source_registry: Path) -> str: + return configuration_payload_revision( + {"source_registry": str(source_registry.expanduser().resolve())} + ).removeprefix("sha256:") + + +def _projected_source_identity( + target_registry: Path, + *, + goal_id: str, +) -> str | None: + target_goal = _goal_or_none(load_registry(target_registry), goal_id) + if target_goal is None: + return None + source_ref = str(target_goal.get("source_registry") or "").strip() + if not source_ref: + return None + return _goal_activation_source_identity(Path(source_ref)) + + +def goal_activation_source_fingerprint( + *, + goal_id: str, + source_registry: Path, + source_bytes: bytes, +) -> str: + identity = { + "schema_version": GOAL_ACTIVATION_SOURCE_FINGERPRINT_SCHEMA_VERSION, + "goal_id": goal_id, + "source_identity": _goal_activation_source_identity(source_registry), + "source_content_sha256": hashlib.sha256(source_bytes).hexdigest(), + } + return configuration_payload_revision(identity).removeprefix("sha256:") + + def _goal(payload: dict[str, Any], goal_id: str) -> dict[str, Any]: goal = next( ( @@ -231,13 +272,28 @@ def set_goal_activation_state( source_registry = authority_route.source_registry target_registry = authority_route.target_registry sync_runtime_root = authority_route.sync_runtime_root + registries_are_distinct = not _same_path(source_registry, target_registry) + source_identity = _goal_activation_source_identity(source_registry) + target_source_identity = ( + _projected_source_identity( + target_registry, + goal_id=normalized_goal_id, + ) + if registries_are_distinct + else source_identity + ) + source_bytes = source_registry.read_bytes() source_goal = _goal(load_registry(source_registry), normalized_goal_id) normalized_fingerprint = str(expected_state_fingerprint or "").strip() or None if normalized_fingerprint is not None and not _SHA256.fullmatch( normalized_fingerprint ): raise ValueError("expected state fingerprint must be a SHA-256 digest") - observed_fingerprint = hashlib.sha256(source_registry.read_bytes()).hexdigest() + observed_fingerprint = goal_activation_source_fingerprint( + goal_id=normalized_goal_id, + source_registry=source_registry, + source_bytes=source_bytes, + ) before_state = goal_activation_state(source_goal) changed = before_state is not target_state actor_label = actor.value if actor is not None else "owner" @@ -265,6 +321,10 @@ def set_goal_activation_state( "written": False, "partial_write": False, "source_registry": str(source_registry), + "source_identity": source_identity, + "source_fingerprint_schema_version": ( + GOAL_ACTIVATION_SOURCE_FINGERPRINT_SCHEMA_VERSION + ), "target_global_registry": str(target_registry), "authority_route": authority_route.public_summary(), "expected_state_fingerprint": normalized_fingerprint, @@ -300,7 +360,6 @@ def set_goal_activation_state( "Repair the Goal source registry route before resuming this Goal." ) - registries_are_distinct = not _same_path(source_registry, target_registry) if registries_are_distinct: writability = probe_registry_write_path(target_registry, create_parent=True) payload["global_registry_writability"] = writability @@ -318,71 +377,124 @@ def set_goal_activation_state( ) return payload - if changed: - with project_registry_transaction( - source_registry, - operation="set_goal_activation_state", - ) as transaction: - source_payload = transaction.payload_copy() - locked_fingerprint = hashlib.sha256(source_registry.read_bytes()).hexdigest() - if ( - normalized_fingerprint is not None - and locked_fingerprint != normalized_fingerprint - ): - payload.update( + sync_payload: dict[str, Any] | None = None + registry_paths = [source_registry] + if registries_are_distinct: + registry_paths.append(target_registry) + with ExitStack() as stack: + transaction = None + for path in sorted(registry_paths, key=lambda item: str(item)): + if _same_path(path, source_registry): + transaction = stack.enter_context( + project_registry_transaction( + source_registry, + operation="set_goal_activation_state", + ) + ) + else: + stack.enter_context( + exclusive_file_lock( + target_registry, + operation="set_goal_activation_state", + ) + ) + if transaction is None: + raise RuntimeError("Goal source registry transaction was not acquired") + if registries_are_distinct: + locked_target_source_identity = _projected_source_identity( + target_registry, + goal_id=normalized_goal_id, + ) + route_changed = locked_target_source_identity != target_source_identity + route_conflicts = locked_target_source_identity not in { + None, + source_identity, + } + if route_changed or route_conflicts: + route_fingerprint = configuration_payload_revision( { - "ok": False, - "error_kind": "goal_action_stale", - "error": ( - "Goal state changed after action projection; refresh actions and retry" - ), - "observed_state_fingerprint": locked_fingerprint, + "source_fingerprint": observed_fingerprint, + "target_source_identity": locked_target_source_identity, } - ) - return payload - locked_goal = _goal(source_payload, normalized_goal_id) - locked_state = goal_activation_state(locked_goal) - if locked_state is not before_state: + ).removeprefix("sha256:") payload.update( { "ok": False, - "error_kind": "goal_activation_state_changed", + "error_kind": "goal_action_stale", "error": ( - "goal activation changed after preview; regenerate the transition" + "Goal source route changed after action projection; " + "refresh actions and retry" ), - "observed_state": locked_state.value, + "observed_state_fingerprint": route_fingerprint, } ) return payload + + source_payload = transaction.payload_copy() + locked_fingerprint = goal_activation_source_fingerprint( + goal_id=normalized_goal_id, + source_registry=source_registry, + source_bytes=source_registry.read_bytes(), + ) + if ( + normalized_fingerprint is not None + and locked_fingerprint != normalized_fingerprint + ): + payload.update( + { + "ok": False, + "error_kind": "goal_action_stale", + "error": ( + "Goal state changed after action projection; refresh actions and retry" + ), + "observed_state_fingerprint": locked_fingerprint, + } + ) + return payload + locked_goal = _goal(source_payload, normalized_goal_id) + locked_state = goal_activation_state(locked_goal) + if locked_state is not before_state: + payload.update( + { + "ok": False, + "error_kind": "goal_activation_state_changed", + "error": ( + "goal activation changed after preview; regenerate the transition" + ), + "observed_state": locked_state.value, + } + ) + return payload + if changed: locked_goal.pop("activation_state", None) locked_goal["activation"] = proposed_activation transaction.commit(source_payload) payload["written"] = True - sync_payload: dict[str, Any] | None = None - if registries_are_distinct: - sync_payload = sync_project_registry_to_global( - registry_path=source_registry, - runtime_root_override=sync_runtime_root, - goal_id=normalized_goal_id, - dry_run=False, - ) - payload["global_sync"] = sync_payload + if registries_are_distinct: + sync_payload = sync_project_registry_to_global( + registry_path=source_registry, + runtime_root_override=sync_runtime_root, + goal_id=normalized_goal_id, + dry_run=False, + _global_registry_lock_held=True, + ) + payload["global_sync"] = sync_payload - readback = ( - { - "schema_version": GOAL_ACTIVATION_READBACK_SCHEMA_VERSION, - "status": "not_run", - "verified": False, - } - if sync_payload is not None and not sync_payload.get("ok") - else _readback( - source_registry=source_registry, - target_registry=target_registry, - goal_id=normalized_goal_id, - expected_state=target_state, + readback = ( + { + "schema_version": GOAL_ACTIVATION_READBACK_SCHEMA_VERSION, + "status": "not_run", + "verified": False, + } + if sync_payload is not None and not sync_payload.get("ok") + else _readback( + source_registry=source_registry, + target_registry=target_registry, + goal_id=normalized_goal_id, + expected_state=target_state, + ) ) - ) payload["readback"] = readback sync_ok = sync_payload is None or bool(sync_payload.get("ok")) payload["ok"] = bool(sync_ok and readback.get("verified")) diff --git a/loopx/control_plane/goals/operator_actions.py b/loopx/control_plane/goals/operator_actions.py index 09bc1b55fd..3def6e99ca 100644 --- a/loopx/control_plane/goals/operator_actions.py +++ b/loopx/control_plane/goals/operator_actions.py @@ -1,7 +1,6 @@ from __future__ import annotations from collections.abc import Mapping -import hashlib from pathlib import Path from typing import Any @@ -14,6 +13,7 @@ GoalActivationAuthorityRouteMode, GoalActivationSourceStatus, _source_and_target, + goal_activation_source_fingerprint, ) @@ -118,7 +118,11 @@ def build_goal_action_catalog( if authority_route.source_status is GoalActivationSourceStatus.AVAILABLE else None ) - fingerprint = hashlib.sha256(source_bytes).hexdigest() + fingerprint = goal_activation_source_fingerprint( + goal_id=normalized_goal_id, + source_registry=authority_route.source_registry, + source_bytes=source_bytes, + ) try: result = effect_runtime_result( "goal.operator_actions.project", diff --git a/loopx/global_registry.py b/loopx/global_registry.py index 42c36243bd..d2dde65a0b 100644 --- a/loopx/global_registry.py +++ b/loopx/global_registry.py @@ -58,34 +58,30 @@ def _global_registry_backup_path(global_path: Path, label: str) -> Path: return global_path.with_name(f"{global_path.name}.{label}-{timestamp}.bak") -def mutate_global_registry( +def _mutate_global_registry_locked( global_path: Path, - operation: str, reducer: Callable[[dict[str, Any]], GlobalRegistryReduction], ) -> dict[str, Any]: - """Apply one authoritative global-registry read-modify-write transaction.""" + current = _load_global_registry(global_path) + reduction = reducer(copy.deepcopy(current)) + if not isinstance(reduction, GlobalRegistryReduction): + raise TypeError( + "global registry reducer must return GlobalRegistryReduction" + ) + if not isinstance(reduction.payload, dict): + raise TypeError("global registry reducer payload must be a JSON object") - with exclusive_file_lock(global_path, operation=operation): - current = _load_global_registry(global_path) - reduction = reducer(copy.deepcopy(current)) - if not isinstance(reduction, GlobalRegistryReduction): - raise TypeError( - "global registry reducer must return GlobalRegistryReduction" - ) - if not isinstance(reduction.payload, dict): - raise TypeError("global registry reducer payload must be a JSON object") - - wrote = reduction.payload != current - backup_path = None - if wrote and reduction.backup_label and global_path.exists(): - backup = _global_registry_backup_path( - global_path, - reduction.backup_label, - ) - write_json(backup, current) - backup_path = str(backup) - if wrote: - write_json(global_path, reduction.payload) + wrote = reduction.payload != current + backup_path = None + if wrote and reduction.backup_label and global_path.exists(): + backup = _global_registry_backup_path( + global_path, + reduction.backup_label, + ) + write_json(backup, current) + backup_path = str(backup) + if wrote: + write_json(global_path, reduction.payload) return { "before": current, @@ -96,6 +92,17 @@ def mutate_global_registry( } +def mutate_global_registry( + global_path: Path, + operation: str, + reducer: Callable[[dict[str, Any]], GlobalRegistryReduction], +) -> dict[str, Any]: + """Apply one authoritative global-registry read-modify-write transaction.""" + + with exclusive_file_lock(global_path, operation=operation): + return _mutate_global_registry_locked(global_path, reducer) + + def global_write_denied_payload( *, registry_path: Path, @@ -631,6 +638,7 @@ def sync_project_registry_to_global( goal_id: str | None = None, dry_run: bool = False, allow_route_replacement: bool = False, + _global_registry_lock_held: bool = False, ) -> dict[str, Any]: registry_path = registry_path.expanduser() if not registry_path.exists(): @@ -744,15 +752,25 @@ def sync_project_registry_to_global( else None ) else: - mutation = mutate_global_registry( - global_path, - "sync_global_registry", - lambda current: _sync_global_registry_reduction( + def reduce_current(current: dict[str, Any]) -> GlobalRegistryReduction: + return _sync_global_registry_reduction( current, incoming, incoming_projects, **merge_kwargs, - ), + ) + + mutation = ( + _mutate_global_registry_locked( + global_path, + reduce_current, + ) + if _global_registry_lock_held + else mutate_global_registry( + global_path, + "sync_global_registry", + reduce_current, + ) ) receipt = mutation["receipt"] merged_goals = receipt["merged_goals"] diff --git a/tests/control_plane/test_goal_activation.py b/tests/control_plane/test_goal_activation.py index c54ac09a72..5941d8979b 100644 --- a/tests/control_plane/test_goal_activation.py +++ b/tests/control_plane/test_goal_activation.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import contextmanager import json from pathlib import Path @@ -8,6 +9,7 @@ from loopx import chat_goal_lifecycle_actions from loopx.chat_action_store import ChatActionStore from loopx.chat_actions import ChatActionService +from loopx.control_plane.goals import activation_service from loopx.control_plane.goals.activation import ( GoalActivationState, build_goal_activation, @@ -127,6 +129,10 @@ def test_stop_preview_is_zero_write(connected_registries: tuple[Path, Path]) -> assert result["dry_run"] is True assert result["changed"] is True assert result["written"] is False + assert ( + result["source_fingerprint_schema_version"] + == "loopx_goal_activation_source_fingerprint_v1" + ) assert source_registry.read_bytes() == before_source assert global_registry.read_bytes() == before_global @@ -475,6 +481,49 @@ def test_owner_confirmed_lifecycle_action_rejects_a_changed_source_registry( ) +def test_owner_confirmed_lifecycle_action_rejects_equal_content_source_route_change( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + source_a, global_registry = connected_registries + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Stop a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "stop", + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "stop-goal-one-before-source-route-change", + } + ) + source_b = tmp_path / "project-b" / ".loopx" / "registry.json" + source_b.parent.mkdir(parents=True) + source_b.write_bytes(source_a.read_bytes()) + global_payload = load_registry(global_registry) + registry_goals(global_payload)[0]["source_registry"] = str(source_b) + _write_json(global_registry, global_payload) + before = source_a.read_bytes(), source_b.read_bytes(), global_registry.read_bytes() + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert ( + source_a.read_bytes(), + source_b.read_bytes(), + global_registry.read_bytes(), + ) == before + assert goal_activation_state(_goal(source_a)) is GoalActivationState.ACTIVE + assert goal_activation_state(_goal(source_b)) is GoalActivationState.ACTIVE + + def test_owner_confirmed_lifecycle_action_rechecks_source_inside_write_lock( connected_registries: tuple[Path, Path], tmp_path: Path, @@ -532,6 +581,161 @@ def change_source_before_execute(**kwargs: object) -> dict[str, object]: ) +def test_owner_confirmed_lifecycle_action_rechecks_route_before_execute( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_a, global_registry = connected_registries + source_b = tmp_path / "project-b" / ".loopx" / "registry.json" + source_b.parent.mkdir(parents=True) + source_b.write_bytes(source_a.read_bytes()) + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Stop a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "stop", + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "stop-goal-one-before-execute-route-race", + } + ) + original = chat_goal_lifecycle_actions.set_goal_activation_state + switched_global_bytes: bytes | None = None + + def change_route_before_execute(**kwargs: object) -> dict[str, object]: + nonlocal switched_global_bytes + if kwargs.get("execute") is True and switched_global_bytes is None: + global_payload = load_registry(global_registry) + registry_goals(global_payload)[0]["source_registry"] = str(source_b) + _write_json(global_registry, global_payload) + switched_global_bytes = global_registry.read_bytes() + return original(**kwargs) + + monkeypatch.setattr( + chat_goal_lifecycle_actions, + "set_goal_activation_state", + change_route_before_execute, + ) + before_sources = source_a.read_bytes(), source_b.read_bytes() + + applied = service.apply(str(proposal["proposal_id"])) + + assert switched_global_bytes is not None + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert (source_a.read_bytes(), source_b.read_bytes()) == before_sources + assert global_registry.read_bytes() == switched_global_bytes + assert goal_activation_state(_goal(source_a)) is GoalActivationState.ACTIVE + assert goal_activation_state(_goal(source_b)) is GoalActivationState.ACTIVE + + +def test_lifecycle_execute_rechecks_route_before_source_commit( + connected_registries: tuple[Path, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_a, global_registry = connected_registries + preview = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state=GoalActivationState.STOPPED, + execute=False, + ) + source_b = tmp_path / "project-b" / ".loopx" / "registry.json" + source_b.parent.mkdir(parents=True) + source_b.write_bytes(source_a.read_bytes()) + original_transaction = activation_service.project_registry_transaction + switched_global_bytes: bytes | None = None + + @contextmanager + def change_route_before_source_lock(*args: object, **kwargs: object): + nonlocal switched_global_bytes + global_payload = load_registry(global_registry) + registry_goals(global_payload)[0]["source_registry"] = str(source_b) + _write_json(global_registry, global_payload) + switched_global_bytes = global_registry.read_bytes() + with original_transaction(*args, **kwargs) as transaction: + yield transaction + + monkeypatch.setattr( + activation_service, + "project_registry_transaction", + change_route_before_source_lock, + ) + before_sources = source_a.read_bytes(), source_b.read_bytes() + + result = set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state=GoalActivationState.STOPPED, + actor_kind="owner", + expected_state_fingerprint=preview["observed_state_fingerprint"], + execute=True, + ) + + assert switched_global_bytes is not None + assert result["ok"] is False + assert result["error_kind"] == "goal_action_stale" + assert result["written"] is False + assert (source_a.read_bytes(), source_b.read_bytes()) == before_sources + assert global_registry.read_bytes() == switched_global_bytes + + +def test_owner_confirmed_lifecycle_action_does_not_recover_across_route_change( + connected_registries: tuple[Path, Path], + tmp_path: Path, +) -> None: + source_a, global_registry = connected_registries + service = ChatActionService( + store=ChatActionStore(tmp_path / "actions"), + registry_path=global_registry, + ) + proposal = service.preview( + { + "action_kind": "goal.lifecycle", + "summary": "Stop a Goal", + "normalized_parameters": { + "goal_id": "goal-one", + "operation": "stop", + "reason": "Owner confirmed from the workspace", + }, + "context": {"kind": "goal_directory"}, + "idempotency_key": "stop-goal-one-before-route-recovery", + } + ) + source_b = tmp_path / "project-b" / ".loopx" / "registry.json" + source_b_payload = load_registry(source_a) + registry_goals(source_b_payload)[0]["activation"] = build_goal_activation( + state=GoalActivationState.STOPPED, + updated_at="2026-09-24T00:00:00+00:00", + reason="Stopped through another source", + actor_kind="owner", + ) + _write_json(source_b, source_b_payload) + global_payload = load_registry(global_registry) + registry_goals(global_payload)[0]["source_registry"] = str(source_b) + _write_json(global_registry, global_payload) + before = source_a.read_bytes(), source_b.read_bytes(), global_registry.read_bytes() + + applied = service.apply(str(proposal["proposal_id"])) + + assert applied["proposal"]["status"] == "stale" + assert applied["proposal"]["receipt"] is None + assert ( + source_a.read_bytes(), + source_b.read_bytes(), + global_registry.read_bytes(), + ) == before + + def test_owner_confirmed_lifecycle_action_recovers_after_committed_stop( connected_registries: tuple[Path, Path], tmp_path: Path, From ac89af887486946ca372effde87652464674c773 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Thu, 24 Sep 2026 16:41:09 +0800 Subject: [PATCH 3/3] fix: align goal activation registry lock order Signed-off-by: duanjialing.777 --- .../control_plane/goals/activation_service.py | 193 +++++++++--------- tests/control_plane/test_goal_activation.py | 99 +++++++++ 2 files changed, 192 insertions(+), 100 deletions(-) diff --git a/loopx/control_plane/goals/activation_service.py b/loopx/control_plane/goals/activation_service.py index d0ece3f5d3..8599146b7c 100644 --- a/loopx/control_plane/goals/activation_service.py +++ b/loopx/control_plane/goals/activation_service.py @@ -1,6 +1,6 @@ from __future__ import annotations -from contextlib import ExitStack +from contextlib import nullcontext from dataclasses import dataclass from enum import Enum import hashlib @@ -378,123 +378,116 @@ def set_goal_activation_state( return payload sync_payload: dict[str, Any] | None = None - registry_paths = [source_registry] - if registries_are_distinct: - registry_paths.append(target_registry) - with ExitStack() as stack: - transaction = None - for path in sorted(registry_paths, key=lambda item: str(item)): - if _same_path(path, source_registry): - transaction = stack.enter_context( - project_registry_transaction( - source_registry, - operation="set_goal_activation_state", - ) + with project_registry_transaction( + source_registry, + operation="set_goal_activation_state", + ) as transaction: + target_lock = ( + exclusive_file_lock( + target_registry, + operation="set_goal_activation_state", + ) + if registries_are_distinct + else nullcontext() + ) + with target_lock: + if registries_are_distinct: + locked_target_source_identity = _projected_source_identity( + target_registry, + goal_id=normalized_goal_id, ) - else: - stack.enter_context( - exclusive_file_lock( - target_registry, - operation="set_goal_activation_state", + route_changed = locked_target_source_identity != target_source_identity + route_conflicts = locked_target_source_identity not in { + None, + source_identity, + } + if route_changed or route_conflicts: + route_fingerprint = configuration_payload_revision( + { + "source_fingerprint": observed_fingerprint, + "target_source_identity": locked_target_source_identity, + } + ).removeprefix("sha256:") + payload.update( + { + "ok": False, + "error_kind": "goal_action_stale", + "error": ( + "Goal source route changed after action projection; " + "refresh actions and retry" + ), + "observed_state_fingerprint": route_fingerprint, + } ) - ) - if transaction is None: - raise RuntimeError("Goal source registry transaction was not acquired") - if registries_are_distinct: - locked_target_source_identity = _projected_source_identity( - target_registry, + return payload + + source_payload = transaction.payload_copy() + locked_fingerprint = goal_activation_source_fingerprint( goal_id=normalized_goal_id, + source_registry=source_registry, + source_bytes=source_registry.read_bytes(), ) - route_changed = locked_target_source_identity != target_source_identity - route_conflicts = locked_target_source_identity not in { - None, - source_identity, - } - if route_changed or route_conflicts: - route_fingerprint = configuration_payload_revision( - { - "source_fingerprint": observed_fingerprint, - "target_source_identity": locked_target_source_identity, - } - ).removeprefix("sha256:") + if ( + normalized_fingerprint is not None + and locked_fingerprint != normalized_fingerprint + ): payload.update( { "ok": False, "error_kind": "goal_action_stale", "error": ( - "Goal source route changed after action projection; " + "Goal state changed after action projection; " "refresh actions and retry" ), - "observed_state_fingerprint": route_fingerprint, + "observed_state_fingerprint": locked_fingerprint, } ) return payload + locked_goal = _goal(source_payload, normalized_goal_id) + locked_state = goal_activation_state(locked_goal) + if locked_state is not before_state: + payload.update( + { + "ok": False, + "error_kind": "goal_activation_state_changed", + "error": ( + "goal activation changed after preview; " + "regenerate the transition" + ), + "observed_state": locked_state.value, + } + ) + return payload + if changed: + locked_goal.pop("activation_state", None) + locked_goal["activation"] = proposed_activation + transaction.commit(source_payload) + payload["written"] = True + + if registries_are_distinct: + sync_payload = sync_project_registry_to_global( + registry_path=source_registry, + runtime_root_override=sync_runtime_root, + goal_id=normalized_goal_id, + dry_run=False, + _global_registry_lock_held=True, + ) + payload["global_sync"] = sync_payload - source_payload = transaction.payload_copy() - locked_fingerprint = goal_activation_source_fingerprint( - goal_id=normalized_goal_id, - source_registry=source_registry, - source_bytes=source_registry.read_bytes(), - ) - if ( - normalized_fingerprint is not None - and locked_fingerprint != normalized_fingerprint - ): - payload.update( - { - "ok": False, - "error_kind": "goal_action_stale", - "error": ( - "Goal state changed after action projection; refresh actions and retry" - ), - "observed_state_fingerprint": locked_fingerprint, - } - ) - return payload - locked_goal = _goal(source_payload, normalized_goal_id) - locked_state = goal_activation_state(locked_goal) - if locked_state is not before_state: - payload.update( + readback = ( { - "ok": False, - "error_kind": "goal_activation_state_changed", - "error": ( - "goal activation changed after preview; regenerate the transition" - ), - "observed_state": locked_state.value, + "schema_version": GOAL_ACTIVATION_READBACK_SCHEMA_VERSION, + "status": "not_run", + "verified": False, } + if sync_payload is not None and not sync_payload.get("ok") + else _readback( + source_registry=source_registry, + target_registry=target_registry, + goal_id=normalized_goal_id, + expected_state=target_state, + ) ) - return payload - if changed: - locked_goal.pop("activation_state", None) - locked_goal["activation"] = proposed_activation - transaction.commit(source_payload) - payload["written"] = True - - if registries_are_distinct: - sync_payload = sync_project_registry_to_global( - registry_path=source_registry, - runtime_root_override=sync_runtime_root, - goal_id=normalized_goal_id, - dry_run=False, - _global_registry_lock_held=True, - ) - payload["global_sync"] = sync_payload - - readback = ( - { - "schema_version": GOAL_ACTIVATION_READBACK_SCHEMA_VERSION, - "status": "not_run", - "verified": False, - } - if sync_payload is not None and not sync_payload.get("ok") - else _readback( - source_registry=source_registry, - target_registry=target_registry, - goal_id=normalized_goal_id, - expected_state=target_state, - ) - ) payload["readback"] = readback sync_ok = sync_payload is None or bool(sync_payload.get("ok")) payload["ok"] = bool(sync_ok and readback.get("verified")) diff --git a/tests/control_plane/test_goal_activation.py b/tests/control_plane/test_goal_activation.py index 5941d8979b..3997a09a19 100644 --- a/tests/control_plane/test_goal_activation.py +++ b/tests/control_plane/test_goal_activation.py @@ -1,14 +1,17 @@ from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager import json from pathlib import Path +import threading import pytest from loopx import chat_goal_lifecycle_actions from loopx.chat_action_store import ChatActionStore from loopx.chat_actions import ChatActionService +from loopx.cli_commands import registry_admin from loopx.control_plane.goals import activation_service from loopx.control_plane.goals.activation import ( GoalActivationState, @@ -272,6 +275,102 @@ def test_stop_and_resume_sync_source_global_and_quota( assert resumed_quota["allowed_slots"] == 4 +def test_activation_and_agent_registration_share_source_to_global_lock_order( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "a-runtime" + source_registry = tmp_path / "z-project" / ".loopx" / "registry.json" + source_payload: dict[str, object] = { + "schema_version": "0.1", + "common_runtime_root": str(runtime_root), + "goals": [ + { + "id": "goal-one", + "display_name": "A public Goal", + "repo": str(source_registry.parent.parent), + "quota": {"compute": 1, "allowed_slots": 4, "spent_slots": 0}, + "coordination": { + "registered_agents": ["codex-existing"], + "agent_model": "peer_v1", + }, + } + ], + } + _write_json(source_registry, source_payload) + synced = sync_project_registry_to_global( + registry_path=source_registry, + runtime_root_override=str(runtime_root), + goal_id="goal-one", + dry_run=False, + ) + assert synced["ok"] is True + global_registry = runtime_root / "registry.global.json" + assert str(global_registry) < str(source_registry) + + source_lock_held = threading.Event() + activation_source_lock_attempted = threading.Event() + original_configure_goal = registry_admin.configure_goal + original_project_registry_transaction = ( + activation_service.project_registry_transaction + ) + + def delayed_configure_goal(*args: object, **kwargs: object) -> dict[str, object]: + source_lock_held.set() + assert activation_source_lock_attempted.wait(timeout=5) + return original_configure_goal(*args, **kwargs) + + @contextmanager + def observed_activation_source_transaction(*args: object, **kwargs: object): + activation_source_lock_attempted.set() + with original_project_registry_transaction(*args, **kwargs) as transaction: + yield transaction + + monkeypatch.setattr(registry_admin, "configure_goal", delayed_configure_goal) + monkeypatch.setattr( + activation_service, + "project_registry_transaction", + observed_activation_source_transaction, + ) + + def register_agent() -> dict[str, object]: + return registry_admin.register_agent_via_source_registry( + runtime_root_arg=str(runtime_root), + goal_id="goal-one", + agent_ids=["codex-fresh"], + require_new=True, + execute=True, + ) + + def stop_goal() -> dict[str, object]: + return set_goal_activation_state( + registry_path=global_registry, + goal_id="goal-one", + state="stopped", + actor_kind="owner", + execute=True, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + registration_future = executor.submit(register_agent) + assert source_lock_held.wait(timeout=5) + activation_future = executor.submit(stop_goal) + registration = registration_future.result(timeout=10) + activation = activation_future.result(timeout=10) + + assert registration["ok"] is True + assert registration["registration_readback"]["verified"] is True + assert activation["ok"] is True + assert activation["readback"]["verified"] is True + for registry in (source_registry, global_registry): + goal = _goal(registry) + assert goal_activation_state(goal) is GoalActivationState.STOPPED + assert goal["coordination"]["registered_agents"] == [ + "codex-existing", + "codex-fresh", + ] + + def test_stopped_goal_and_zero_compute_keep_distinct_resume_authority() -> None: stopped_status = quota_status_payload( goal_id="goal-one",