Add typed Goal lifecycle action catalog

loopx/cli_commands/goal_actions.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import argparse
+from collections.abc import Callable
+from pathlib import Path
+
+from ..control_plane.goals.operator_actions import (
+    GOAL_ACTION_CATALOG_SCHEMA_VERSION,
+    build_goal_action_catalog,
+    render_goal_action_catalog_markdown,
+)
+
+
+PrintPayload = Callable[
+    [dict[str, object], str, Callable[[dict[str, object]], str]],
+    None,
+]
+
+
+def register_goal_actions_command(
+    subparsers: argparse._SubParsersAction[argparse.ArgumentParser],
+) -> None:
+    parser = subparsers.add_parser(
+        "goal-actions",
+        help="List fresh typed owner actions for one Goal.",
+    )
+    parser.add_argument(
+        "--goal-id", required=True, help="Goal id present in the active registry."
+    )
+
+
+def handle_goal_actions_command(
+    args: argparse.Namespace,
+    *,
+    registry_path: Path,
+    print_payload: PrintPayload,
+) -> int:
+    try:
+        payload = build_goal_action_catalog(
+            registry_path=registry_path,
+            goal_id=args.goal_id,
+            runtime_root_override=args.runtime_root,
+        )
+    except Exception as exc:
+        payload = {
+            "ok": False,
+            "schema_version": GOAL_ACTION_CATALOG_SCHEMA_VERSION,
+            "goal_id": args.goal_id,
+            "actions": [],
+            "error": str(exc),
+        }
+    print_payload(payload, args.format, render_goal_action_catalog_markdown)
+    return 0 if payload.get("ok") else 1

loopx/cli_commands/goal_lifecycle.py

@@ -34,6 +34,10 @@ def register_goal_lifecycle_command(
         help="Stop automatic advancement or restore eligibility.",
     )
     parser.add_argument("--reason", help="Bounded owner-visible transition reason.")
+    parser.add_argument(
+        "--expected-state-fingerprint",
+        help="SHA-256 registry fingerprint from a fresh goal-actions projection.",
+    )
     parser.add_argument(
         "--execute",
         action="store_true",
@@ -54,6 +58,7 @@ def handle_goal_lifecycle_command(
             state="stopped" if args.operation == "stop" else "active",
             reason=args.reason,
             runtime_root_override=args.runtime_root,
+            expected_state_fingerprint=args.expected_state_fingerprint,
             execute=bool(args.execute),
         )
     except Exception as exc:

loopx/cli_commands/registry_admin.py

@@ -25,6 +25,7 @@
     handle_goal_lifecycle_command,
     register_goal_lifecycle_command,
 )
+from .goal_actions import handle_goal_actions_command, register_goal_actions_command
 from .registry_admin_configure import register_configure_goal_command
 from .registry_admin_lifecycle import (
     REGISTRY_LIFECYCLE_COMMANDS,
@@ -52,6 +53,7 @@
 REGISTRY_ADMIN_COMMANDS = {
     "configure-goal",
     "goal-lifecycle",
+    "goal-actions",
     "register-agent",
     "resolve-agent-thread",
     "bind-agent-thread",
@@ -360,6 +362,7 @@ def loop_activation_for_goal(
 def register_registry_admin_commands(subparsers: argparse._SubParsersAction) -> None:
     register_configure_goal_command(subparsers)
     register_goal_lifecycle_command(subparsers)
+    register_goal_actions_command(subparsers)
 
     register_agent_parser = subparsers.add_parser(
         "register-agent",
@@ -429,6 +432,13 @@ def handle_registry_admin_command(
             print_payload=print_payload,
         )
 
+    if args.command == "goal-actions":
+        return handle_goal_actions_command(
+            args,
+            registry_path=registry_path,
+            print_payload=print_payload,
+        )
+
     if args.command == "configure-goal":
         try:
             agent_work_modes: dict[str, str] = {}

loopx/control_plane/effect_runtime_handlers.ts

@@ -90,6 +90,7 @@ import { buildVisionCheckpoint } from "./goals/vision_checkpoint.ts";
 import { projectVisionWaitCoverage } from "./goals/vision_wait_coverage.ts";
 import { admitGoalAmendmentProposal } from "./goals/goal_amendment_proposal.ts";
 import { projectSharedGoalAlignment } from "./goals/shared_goal_alignment.ts";
+import { projectGoalOperatorActions } from "./goals/operator_actions.ts";
 import {
   evaluateDeliveryRoute,
 } from "./turn_driver/delivery_continuity.ts";
@@ -445,6 +446,7 @@ export function createEffectRuntimeHandlers(
     ["goal.vision_checkpoint.evaluate", buildVisionCheckpoint],
     ["goal.vision_wait.coverage", projectVisionWaitCoverage],
     ["goal.shared_goal_alignment.project", projectSharedGoalAlignment],
+    ["goal.operator_actions.project", projectGoalOperatorActions],
     ["goal.amendment_proposal.admit", admitGoalAmendmentProposal],
     ["agent.delivery_workspace.evaluate", evaluateDeliveryWorkspace],
     [

loopx/control_plane/goals/activation_service.py

@@ -2,7 +2,9 @@
 
 from dataclasses import dataclass
 from enum import Enum
+import hashlib
 from pathlib import Path
+import re
 from typing import Any
 
 from ...file_lock import exclusive_file_lock
@@ -25,6 +27,7 @@
 GOAL_ACTIVATION_AUTHORITY_ROUTE_SCHEMA_VERSION = (
     "loopx_goal_activation_authority_route_v1"
 )
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
 
 
 class GoalActivationAuthorityRouteMode(str, Enum):
@@ -207,6 +210,7 @@ def set_goal_activation_state(
     state: GoalActivationState | str,
     reason: str | None = None,
     runtime_root_override: str | None = None,
+    expected_state_fingerprint: str | None = None,
     execute: bool = False,
 ) -> dict[str, Any]:
     """Preview or apply one reversible Goal activation transition."""
@@ -225,6 +229,12 @@ def set_goal_activation_state(
     target_registry = authority_route.target_registry
     sync_runtime_root = authority_route.sync_runtime_root
     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()
     before_state = goal_activation_state(source_goal)
     changed = before_state is not target_state
     default_reason = (
@@ -252,13 +262,27 @@ def set_goal_activation_state(
         "source_registry": str(source_registry),
         "target_global_registry": str(target_registry),
         "authority_route": authority_route.public_summary(),
+        "expected_state_fingerprint": normalized_fingerprint,
+        "observed_state_fingerprint": observed_fingerprint,
         "activation": proposed_activation,
         "readback": {
             "schema_version": GOAL_ACTIVATION_READBACK_SCHEMA_VERSION,
             "status": "not_executed" if changed else "not_required",
             "verified": not changed,
         },
     }
+    if (
+        normalized_fingerprint is not None
+        and observed_fingerprint != normalized_fingerprint
+    ):
+        payload.update(
+            {
+                "ok": False,
+                "error_kind": "goal_action_stale",
+                "error": "Goal state changed after action projection; refresh actions and retry",
+            }
+        )
+        return payload
     if not execute:
         return payload
 
@@ -294,6 +318,22 @@ def set_goal_activation_state(
             operation="set_goal_activation_state",
         ):
             source_payload = load_registry(source_registry)
+            locked_fingerprint = hashlib.sha256(source_registry.read_bytes()).hexdigest()
+            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:

loopx/control_plane/goals/operator_actions.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from ...registry import registry_goals
+from .activation import GoalActivationState, goal_activation_state
+from .activation_service import _source_and_target
+
+
+GOAL_ACTION_PROJECTION_REQUEST_SCHEMA_VERSION = (
+    "loopx_goal_action_projection_request_v2"
+)
+GOAL_ACTION_CATALOG_SCHEMA_VERSION = "loopx_goal_action_catalog_v1"
+
+
+def _goal(payload: Mapping[str, Any], goal_id: str) -> Mapping[str, Any]:
+    goal = next(
+        (
+            item
+            for item in registry_goals(dict(payload))
+            if str(item.get("id") or "") == goal_id
+        ),
+        None,
+    )
+    if goal is None:
+        raise ValueError(f"goal id not found in registry: {goal_id}")
+    return goal
+
+
+def build_goal_action_catalog(
+    *,
+    registry_path: Path,
+    goal_id: str,
+    runtime_root_override: str | None = None,
+) -> dict[str, Any]:
+    """Adapt one stable registry snapshot into the TS-owned action catalog."""
+
+    normalized_goal_id = str(goal_id or "").strip()
+    if not normalized_goal_id:
+        raise ValueError("goal id is required")
+    requested_registry = Path(registry_path).expanduser().resolve()
+    requested_payload = json.loads(requested_registry.read_text(encoding="utf-8"))
+    requested_goal = _goal(requested_payload, normalized_goal_id)
+    current_state = goal_activation_state(requested_goal)
+    target_state = (
+        GoalActivationState.STOPPED
+        if current_state is GoalActivationState.ACTIVE
+        else GoalActivationState.ACTIVE
+    )
+    authority_route = _source_and_target(
+        registry_path=requested_registry,
+        goal_id=normalized_goal_id,
+        target_state=target_state,
+        runtime_root_override=runtime_root_override,
+    )
+    source_bytes = authority_route.source_registry.read_bytes()
+    source_payload = json.loads(source_bytes)
+    source_state = goal_activation_state(_goal(source_payload, normalized_goal_id))
+    fingerprint = hashlib.sha256(source_bytes).hexdigest()
+    try:
+        result = effect_runtime_result(
+            "goal.operator_actions.project",
+            {
+                "schema_version": GOAL_ACTION_PROJECTION_REQUEST_SCHEMA_VERSION,
+                "goal_id": normalized_goal_id,
+                "registry_locator": str(requested_registry),
+                "runtime_root_locator": authority_route.sync_runtime_root,
+                "activation_state": source_state.value,
+                "state_fingerprint": fingerprint,
+            },
+        )
+    except EffectRuntimeRejected as exc:
+        raise ValueError(str(exc)) from None
+    if not isinstance(result, Mapping) or (
+        result.get("schema_version") != GOAL_ACTION_CATALOG_SCHEMA_VERSION
+    ):
+        raise RuntimeError("TypeScript Goal action catalog shape mismatch")
+    actions = result.get("actions")
+    if not isinstance(actions, list) or not all(
+        isinstance(item, Mapping) for item in actions
+    ):
+        raise RuntimeError("TypeScript Goal action list shape mismatch")
+    return dict(result)
+
+
+def render_goal_action_catalog_markdown(payload: dict[str, Any]) -> str:
+    lines = [
+        "# Goal Actions",
+        "",
+        f"- ok: `{str(payload.get('ok')).lower()}`",
+        f"- goal: `{payload.get('goal_id')}`",
+        f"- activation_state: `{payload.get('activation_state')}`",
+    ]
+    actions = payload.get("actions")
+    if isinstance(actions, list):
+        lines.extend(["", "## Available actions", ""])
+        for action in actions:
+            if isinstance(action, Mapping):
+                lines.append(
+                    f"- `{action.get('action_id')}` — {action.get('label')}"
+                )
+    if payload.get("error"):
+        lines.extend(["", f"Error: {payload.get('error')}"])
+    return "\n".join(lines).rstrip() + "\n"