feat: track enabled and disabled runtime targets

actions/publish-runtime-target-lifecycle/action.yml

@@ -0,0 +1,72 @@
+name: Publish runtime target lifecycle
+description: Build and sync one bounded, no-order runtime-target lifecycle snapshot.
+
+inputs:
+  source-id:
+    description: Stable, non-sensitive source identity for this exact target.
+    required: true
+  target-id:
+    description: Stable, non-sensitive target identity.
+    required: true
+  platform:
+    description: Exact platform identifier.
+    required: true
+  configured-state:
+    description: enabled or disabled; this action never changes it.
+    required: true
+  execution-mode:
+    description: Intended dry_run, paper, or live lane.
+    required: true
+  runtime-guard:
+    description: Sanitized runtime-guard result.
+    required: true
+  execution-heartbeat:
+    description: Sanitized execution-heartbeat result.
+    required: true
+  sync-url:
+    description: HTTPS Strategy Switch Console base URL.
+    required: true
+
+runs:
+  using: composite
+  steps:
+    - name: Build and sync bounded target lifecycle
+      shell: bash
+      env:
+        INPUT_SOURCE_ID: ${{ inputs.source-id }}
+        INPUT_TARGET_ID: ${{ inputs.target-id }}
+        INPUT_PLATFORM: ${{ inputs.platform }}
+        INPUT_CONFIGURED_STATE: ${{ inputs.configured-state }}
+        INPUT_EXECUTION_MODE: ${{ inputs.execution-mode }}
+        INPUT_RUNTIME_GUARD: ${{ inputs.runtime-guard }}
+        INPUT_EXECUTION_HEARTBEAT: ${{ inputs.execution-heartbeat }}
+        INPUT_SYNC_URL: ${{ inputs.sync-url }}
+      run: |
+        set -euo pipefail
+        if [[ -z "${EXECUTION_EVIDENCE_SYNC_TOKEN:-}" ]]; then
+          echo "EXECUTION_EVIDENCE_SYNC_TOKEN is required." >&2
+          exit 2
+        fi
+        if [[ ! "$INPUT_SYNC_URL" =~ ^https://[^/?#]+$ ]]; then
+          echo "sync-url must be an HTTPS base URL without a path or query." >&2
+          exit 2
+        fi
+        work_dir="$(mktemp -d)"
+        cleanup() { rm -rf "$work_dir"; }
+        trap cleanup EXIT
+        snapshot_path="$work_dir/runtime-target-lifecycle.json"
+        python3 "$GITHUB_ACTION_PATH/../../python/scripts/runtime_target_lifecycle.py" \
+          --source-id "$INPUT_SOURCE_ID" \
+          --target-id "$INPUT_TARGET_ID" \
+          --platform "$INPUT_PLATFORM" \
+          --configured-state "$INPUT_CONFIGURED_STATE" \
+          --execution-mode "$INPUT_EXECUTION_MODE" \
+          --runtime-guard "$INPUT_RUNTIME_GUARD" \
+          --execution-heartbeat "$INPUT_EXECUTION_HEARTBEAT" \
+          --output "$snapshot_path"
+        curl --fail --silent --show-error \
+          --request POST "${INPUT_SYNC_URL}/api/internal/sync-runtime-target-lifecycle-source" \
+          --header "Authorization: Bearer ${EXECUTION_EVIDENCE_SYNC_TOKEN}" \
+          --header "Content-Type: application/json" \
+          --data-binary "@$snapshot_path" \
+          >/dev/null

docs/runtime_target_lifecycle.md

@@ -0,0 +1,25 @@
+# Runtime target lifecycle snapshot
+
+`qsl_runtime_target_lifecycle_source_snapshot.v1` records the operational
+state of one exact platform target. It is deliberately separate from
+`qsl_execution_evidence_source_snapshot.v1`:
+
+- an **enabled** target continues its runtime guard and execution-heartbeat
+  monitoring;
+- a deliberately **disabled** target remains visible and continues no-order
+  validation, rather than being misreported as an unhealthy execution target;
+- either monitoring failure returns `parked`; it never changes a target's
+  enabled flag, execution mode, credentials, strategy, or order permissions.
+
+Every target record has `no_order: true`. The Worker stores only platform,
+target identity, intended lane, sanitized monitor states, disposition, and
+bounded reason codes. It accepts the same protected
+`EXECUTION_EVIDENCE_SYNC_TOKEN` as the existing platform execution-evidence
+publisher, but stores these snapshots under a separate KV prefix and exposes
+them from `GET /api/runtime-target-lifecycle` only to an allowed signed-in
+user.
+
+Platform workflows call the reusable
+`actions/publish-runtime-target-lifecycle` action after their existing checks.
+The action only constructs and posts a sanitized status object; it has no
+broker SDK, account material, or command to enable a runtime target.

python/scripts/runtime_target_lifecycle.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""Build a bounded, no-order lifecycle snapshot for one runtime target.
+
+This contract intentionally tracks configured target state separately from
+execution evidence.  A target disabled by policy is not an unavailable broker,
+and it must not be represented as a paper/live execution result.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Mapping
+
+
+SOURCE_SCHEMA_VERSION = "qsl_runtime_target_lifecycle_source_snapshot.v1"
+PLATFORMS = frozenset({"alpaca", "binance", "firstrade", "ibkr", "longbridge", "qmt", "schwab"})
+CONFIGURED_STATES = frozenset({"enabled", "disabled"})
+EXECUTION_MODES = frozenset({"dry_run", "paper", "live"})
+CHECK_STATUSES = frozenset({"pass", "attention", "not_due", "not_applicable", "unavailable"})
+DISPOSITIONS = frozenset({"continue_enabled_monitoring", "continue_disabled_validation", "parked"})
+REASON_CODES = frozenset(
+    {
+        "none",
+        "target_intentionally_disabled",
+        "runtime_guard_attention",
+        "execution_heartbeat_attention",
+        "monitoring_unavailable",
+    }
+)
+_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._=-]{0,127}$")
+_TIMESTAMP = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")
+
+
+class RuntimeTargetLifecycleError(ValueError):
+    """Raised when a lifecycle snapshot would be ambiguous or unsafe."""
+
+
+def _identifier(value: object, field: str) -> str:
+    text = str(value or "").strip()
+    if not _IDENTIFIER.fullmatch(text):
+        raise RuntimeTargetLifecycleError(f"{field} must be a stable non-sensitive identifier")
+    return text
+
+
+def _choice(value: object, choices: frozenset[str], field: str) -> str:
+    text = str(value or "").strip()
+    if text not in choices:
+        raise RuntimeTargetLifecycleError(f"{field} is unsupported")
+    return text
+
+
+def _timestamp(value: object | None) -> str:
+    if value is None:
+        return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+    text = str(value).strip()
+    if not _TIMESTAMP.fullmatch(text):
+        raise RuntimeTargetLifecycleError("observed_at must be an RFC3339 UTC timestamp")
+    try:
+        datetime.strptime(text, "%Y-%m-%dT%H:%M:%SZ")
+    except ValueError as exc:
+        raise RuntimeTargetLifecycleError("observed_at must be a valid calendar timestamp") from exc
+    return text
+
+
+def _target_disposition(
+    *,
+    configured_state: str,
+    runtime_guard: str,
+    execution_heartbeat: str,
+) -> tuple[str, str]:
+    if runtime_guard == "attention":
+        return "parked", "runtime_guard_attention"
+    if execution_heartbeat == "attention":
+        return "parked", "execution_heartbeat_attention"
+    if runtime_guard == "unavailable" or execution_heartbeat == "unavailable":
+        return "parked", "monitoring_unavailable"
+    if configured_state == "disabled":
+        if execution_heartbeat != "not_applicable":
+            raise RuntimeTargetLifecycleError(
+                "disabled targets require a not_applicable execution heartbeat"
+            )
+        return "continue_disabled_validation", "target_intentionally_disabled"
+    if runtime_guard not in {"pass", "not_due"} or execution_heartbeat not in {"pass", "not_due"}:
+        raise RuntimeTargetLifecycleError("enabled target monitoring state is incomplete")
+    return "continue_enabled_monitoring", "none"
+
+
+def build_runtime_target_lifecycle_source_snapshot(
+    *,
+    source_id: object,
+    target_id: object,
+    platform: object,
+    configured_state: object,
+    execution_mode: object,
+    runtime_guard: object,
+    execution_heartbeat: object,
+    observed_at: object | None = None,
+) -> dict[str, Any]:
+    """Create one sanitized target state record without execution authority."""
+    normalized_source_id = _identifier(source_id, "source_id")
+    normalized_target_id = _identifier(target_id, "target_id")
+    normalized_platform = _choice(platform, PLATFORMS, "platform")
+    normalized_state = _choice(configured_state, CONFIGURED_STATES, "configured_state")
+    normalized_mode = _choice(execution_mode, EXECUTION_MODES, "execution_mode")
+    normalized_guard = _choice(runtime_guard, CHECK_STATUSES, "runtime_guard")
+    normalized_heartbeat = _choice(execution_heartbeat, CHECK_STATUSES, "execution_heartbeat")
+    disposition, reason_code = _target_disposition(
+        configured_state=normalized_state,
+        runtime_guard=normalized_guard,
+        execution_heartbeat=normalized_heartbeat,
+    )
+    timestamp = _timestamp(observed_at)
+    return {
+        "schema_version": SOURCE_SCHEMA_VERSION,
+        "source_id": normalized_source_id,
+        "generated_at": timestamp,
+        "computed_at": timestamp,
+        "data_status": "ready",
+        "targets": [
+            {
+                "target_id": normalized_target_id,
+                "target": {
+                    "platform": normalized_platform,
+                    "configured_state": normalized_state,
+                    "execution_mode": normalized_mode,
+                },
+                "monitoring": {
+                    "runtime_guard": normalized_guard,
+                    "execution_heartbeat": normalized_heartbeat,
+                },
+                "disposition": {"code": disposition, "reason_code": reason_code},
+                "no_order": True,
+            }
+        ],
+        "errors": [],
+    }
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source-id", required=True)
+    parser.add_argument("--target-id", required=True)
+    parser.add_argument("--platform", required=True)
+    parser.add_argument("--configured-state", required=True, choices=sorted(CONFIGURED_STATES))
+    parser.add_argument("--execution-mode", required=True, choices=sorted(EXECUTION_MODES))
+    parser.add_argument("--runtime-guard", required=True, choices=sorted(CHECK_STATUSES))
+    parser.add_argument("--execution-heartbeat", required=True, choices=sorted(CHECK_STATUSES))
+    parser.add_argument("--observed-at")
+    parser.add_argument("--output", required=True)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(argv)
+    snapshot = build_runtime_target_lifecycle_source_snapshot(
+        source_id=args.source_id,
+        target_id=args.target_id,
+        platform=args.platform,
+        configured_state=args.configured_state,
+        execution_mode=args.execution_mode,
+        runtime_guard=args.runtime_guard,
+        execution_heartbeat=args.execution_heartbeat,
+        observed_at=args.observed_at,
+    )
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

python/tests/test_runtime_target_lifecycle.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).parents[2]
+MODULE_PATH = ROOT / "python" / "scripts" / "runtime_target_lifecycle.py"
+SPEC = importlib.util.spec_from_file_location("runtime_target_lifecycle", MODULE_PATH)
+assert SPEC and SPEC.loader
+lifecycle = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = lifecycle
+SPEC.loader.exec_module(lifecycle)
+
+
+def _snapshot(**overrides: object) -> dict[str, object]:
+    values: dict[str, object] = {
+        "source_id": "longbridge.sg",
+        "target_id": "longbridge.sg",
+        "platform": "longbridge",
+        "configured_state": "disabled",
+        "execution_mode": "dry_run",
+        "runtime_guard": "pass",
+        "execution_heartbeat": "not_applicable",
+        "observed_at": "2026-08-30T00:00:00Z",
+    }
+    values.update(overrides)
+    return lifecycle.build_runtime_target_lifecycle_source_snapshot(**values)
+
+
+class RuntimeTargetLifecycleTest(unittest.TestCase):
+    def test_disabled_target_remains_in_no_order_validation_lane(self) -> None:
+        snapshot = _snapshot()
+
+        target = snapshot["targets"][0]
+        self.assertEqual(
+            snapshot["schema_version"],
+            "qsl_runtime_target_lifecycle_source_snapshot.v1",
+        )
+        self.assertEqual(target["target"]["configured_state"], "disabled")
+        self.assertEqual(
+            target["disposition"],
+            {
+                "code": "continue_disabled_validation",
+                "reason_code": "target_intentionally_disabled",
+            },
+        )
+        self.assertIs(target["no_order"], True)
+
+
+    def test_enabled_target_continues_monitoring_when_checks_pass(self) -> None:
+        snapshot = _snapshot(
+            configured_state="enabled",
+            execution_mode="paper",
+            execution_heartbeat="pass",
+        )
+
+        self.assertEqual(
+            snapshot["targets"][0]["disposition"],
+            {
+                "code": "continue_enabled_monitoring",
+                "reason_code": "none",
+            },
+        )
+
+
+    def test_monitoring_failures_park_without_changing_target_state(self) -> None:
+        cases = [
+        ("attention", "pass", "runtime_guard_attention"),
+        ("pass", "attention", "execution_heartbeat_attention"),
+        ("unavailable", "pass", "monitoring_unavailable"),
+        ]
+        for runtime_guard, execution_heartbeat, reason_code in cases:
+            with self.subTest(
+                runtime_guard=runtime_guard,
+                execution_heartbeat=execution_heartbeat,
+            ):
+                snapshot = _snapshot(
+                    configured_state="enabled",
+                    execution_mode="paper",
+                    runtime_guard=runtime_guard,
+                    execution_heartbeat=execution_heartbeat,
+                )
+
+                self.assertEqual(
+                    snapshot["targets"][0]["disposition"],
+                    {
+                        "code": "parked",
+                        "reason_code": reason_code,
+                    },
+                )
+                self.assertIs(snapshot["targets"][0]["no_order"], True)
+
+
+    def test_disabled_target_rejects_an_execution_heartbeat_claim(self) -> None:
+        with self.assertRaisesRegex(lifecycle.RuntimeTargetLifecycleError, "not_applicable"):
+            _snapshot(execution_heartbeat="pass")