feat: Observe protected plain-DCA PAPER configuration

.github/workflows/inspect-paper-serving-metadata.yml

@@ -12,6 +12,7 @@ on:
           - serving-image
           - source-chain
           - target-config
+          - plain-dca
 
 permissions:
   contents: read
@@ -35,6 +36,10 @@ jobs:
       CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
       CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}
       EXPECTED_RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON || secrets.RUNTIME_TARGET_JSON }}
+      WORKFLOW_ENVIRONMENT: longbridge-paper
+      EXPECTED_PHYSICAL_ACCOUNT_ID: ${{ vars.LONGBRIDGE_PHYSICAL_ACCOUNT_ID || secrets.LONGBRIDGE_PHYSICAL_ACCOUNT_ID }}
+      EXPECTED_RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED || secrets.RUNTIME_TARGET_ENABLED }}
+      EXPECTED_DRY_RUN_ONLY: ${{ vars.LONGBRIDGE_DRY_RUN_ONLY || secrets.LONGBRIDGE_DRY_RUN_ONLY }}
     steps:
       - name: Mask WIF identity components
         env:
@@ -79,6 +84,11 @@ jobs:
         run: |
           python -m pip install --upgrade pip uv
           uv sync --frozen --no-dev
+      - name: Validate plain-DCA protected identity before cloud authentication
+        if: inputs.mode == 'plain-dca'
+        env:
+          GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID }}
+        run: uv run --no-sync python -m scripts.inspect_plain_dca_paper --validate-config
       - uses: google-github-actions/auth@v3
         with:
           workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }}
@@ -93,5 +103,6 @@ jobs:
             serving-image) uv run --no-sync python -m scripts.inspect_paper_serving_metadata ;;
             source-chain) uv run --no-sync python -m scripts.inspect_paper_serving_metadata --source-chain ;;
             target-config) uv run --no-sync python -m scripts.inspect_paper_serving_metadata --target-config ;;
+            plain-dca) uv run --no-sync python -m scripts.inspect_plain_dca_paper ;;
             *) exit 2 ;;
           esac

application/plain_dca_paper_adapter.py

@@ -45,6 +45,8 @@ def _paper_account_from_readback(readback: object, *, now: datetime) -> str:
         manifest = load_runtime_target_manifest()
         entries = [entry for entry in manifest.targets if entry.id == "paper"]
         target = json.loads(env.get("RUNTIME_TARGET_JSON", ""))
+        if "QSL_RUNTIME_TARGET_JSON" in env and json.loads(env["QSL_RUNTIME_TARGET_JSON"]) != target:
+            raise ValueError
     except Exception:
         raise PlainDcaPaperError("paper_target_readback_invalid") from None
     if (len(entries) != 1 or entries[0].environment != PAPER_ENVIRONMENT
@@ -63,7 +65,9 @@ def _paper_account_from_readback(readback: object, *, now: datetime) -> str:
     # False means broker PAPER rather than local dry-run in the shared contract.
     # Phase 1 remains disabled and this module has no order writer in either case.
     if (env.get("LONGBRIDGE_DRY_RUN_ONLY") != "false"
-            or env.get("RUNTIME_TARGET_ENABLED") != "false"):
+            or env.get("RUNTIME_TARGET_ENABLED") != "false"
+            or any(env.get("QSL_" + key, "false") != "false" for key in (
+                "RUNTIME_TARGET_ENABLED", "LONGBRIDGE_DRY_RUN_ONLY"))):
         raise PlainDcaPaperError("paper_disabled_binding_required")
     account = env.get("LONGBRIDGE_PHYSICAL_ACCOUNT_ID")
     if (not isinstance(account, str) or not account.strip() or account != account.strip()

docs/runtime-target-manifest.md

@@ -194,6 +194,8 @@ current protected configuration on every attempt. Its readback contains:
 - `protected_env`: current `CLOUD_RUN_SERVICE`, `RUNTIME_TARGET_JSON`,
   `RUNTIME_TARGET_ENABLED`, `LONGBRIDGE_DRY_RUN_ONLY` and
   `LONGBRIDGE_PHYSICAL_ACCOUNT_ID`, kept in restricted process memory only.
+  Include applicable `QSL_` overrides when present: conflicting or unresolved
+  target, enablement and dry-run aliases are rejected, not silently discarded.
 
 `check_plain_dca_paper_binding(readback, now=now)` checks this contract. It rejects
 `longbridge-sg`, `longbridge-hk`, other target ids, service disagreement with the
@@ -222,3 +224,25 @@ other strategy bindings. It also does not authenticate the readback source or
 credentials: deployed PAPER account/state/schedule isolation, authentic safety
 readback, durable fencing/reconciliation, setup decisions and Phase 2 execution
 remain separate prerequisites. Unit fixtures cannot prove those runtime facts.
+
+### Protected plain-DCA observation
+
+The existing PAPER metadata workflow has a `plain-dca` mode. It uses only the
+fixed `longbridge-paper` Environment and existing protected bindings. Before
+cloud authentication it checks the manifest service/region, explicit PAPER
+selectors, physical-account binding and disabled target. It reads the service
+template and every positive-traffic revision twice, rejecting conflicting
+aliases, duplicate environment keys, missing values and configuration drift.
+It never resolves Secret Manager references, invokes the service or broker, or
+changes cloud configuration. Only fixed diagnostics and comparison results leave
+process memory; the packet is not an adapter admission token.
+
+This is a configuration observation, not independent broker identity proof.
+No authoritative Kelly/all-stop-loss source is implemented in this runtime, so
+those values remain `null`, even if an unrelated flag has a similar name. The
+existing durable live-command switch is reported OFF only when explicitly false
+in both the template and all serving revisions. A disabled PAPER configuration
+does not prove all deployed safety controls. Partial/unknown packets exit with
+failure and `paper_start_ready=false`. Do not create or change protected values
+merely to make this observation pass. A current authoritative safety source and
+independent broker identity evidence remain prerequisites for an owner start card.

scripts/inspect_plain_dca_paper.py

@@ -0,0 +1,166 @@
+"""Read PAPER configuration only; never invoke a service or broker.
+
+This observes configured identity, not credential identity. No authoritative
+Kelly/all-stop-loss reader exists here yet: those fields must remain unknown.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+from datetime import UTC, datetime
+
+from application.runtime_target_manifest import load_runtime_target_manifest
+
+
+class ReadbackError(ValueError):
+    """Only fixed diagnostics may leave protected process memory."""
+
+
+def read_json(args):
+    try:
+        result = subprocess.run(args, capture_output=True, timeout=60, check=False)
+        if result.returncode or len(result.stdout) > 2 * 1024 * 1024:
+            raise ValueError
+        return json.loads(result.stdout)
+    except (OSError, subprocess.TimeoutExpired, ValueError, UnicodeError):
+        raise ReadbackError("protected_read_unavailable") from None
+
+
+def container_env(spec):
+    containers = spec.get("containers")
+    if not isinstance(containers, list) or len(containers) != 1:
+        raise ReadbackError("container_configuration_invalid")
+    entries = containers[0].get("env")
+    if not isinstance(entries, list):
+        raise ReadbackError("container_configuration_invalid")
+    result = {}
+    for entry in entries:
+        if not isinstance(entry, dict) or not isinstance(entry.get("name"), str):
+            raise ReadbackError("container_configuration_invalid")
+        name = entry["name"]
+        if name in result:
+            raise ReadbackError("container_configuration_invalid")
+        # Secret references are deliberately unresolved; never access their values.
+        result[name] = entry.get("value") if "valueFrom" not in entry else None
+    return result
+
+
+def require_identity(values, *, service, account):
+    try:
+        target = json.loads(values["RUNTIME_TARGET_JSON"])
+        alias = values.get("QSL_RUNTIME_TARGET_JSON")
+        if "QSL_RUNTIME_TARGET_JSON" in values and (alias is None or json.loads(alias) != target):
+            raise ValueError
+        required = {"platform_id": "longbridge", "service_name": service,
+                        "execution_environment": "paper", "execution_mode": "live",
+                        "dry_run_only": False, "account_scope": "PAPER",
+                        "account_selector": ["PAPER"], "deployment_selector": "PAPER"}
+        if (not isinstance(target, dict)
+                or any(type(target.get(k)) is not type(v) or target[k] != v for k, v in required.items())
+                or values.get("LONGBRIDGE_PHYSICAL_ACCOUNT_ID") != account
+                or target.get("account_selectors", ["PAPER"]) != ["PAPER"]
+                or target.get("physical_account_id", account) != account
+                or values.get("RUNTIME_TARGET_ENABLED") != "false"
+                or values.get("LONGBRIDGE_DRY_RUN_ONLY") != "false"
+                or any(values.get("QSL_" + key, "false") != "false" for key in (
+                    "RUNTIME_TARGET_ENABLED", "LONGBRIDGE_DRY_RUN_ONLY"))):
+            raise ValueError
+    except (KeyError, TypeError, ValueError):
+        raise ReadbackError("paper_disabled_identity_unproven") from None
+
+
+def configuration(env):
+    paper, = [t for t in load_runtime_target_manifest().targets if t.id == "paper"]
+    service, region, project, account = (env.get(k, "") for k in (
+        "CLOUD_RUN_SERVICE", "CLOUD_RUN_REGION", "GCP_PROJECT_ID", "EXPECTED_PHYSICAL_ACCOUNT_ID"))
+    if (env.get("WORKFLOW_ENVIRONMENT") != "longbridge-paper"
+            or paper.environment != "longbridge-paper" or paper.mode != "paper"
+            or service != paper.service or region != paper.region
+            or not re.fullmatch(r"[a-z][a-z0-9-]{4,61}[a-z0-9]", project)
+            or not account or account != account.strip()
+            or account.upper() in {"PAPER", "LIVE", "SG", "HK", "US", "DEFAULT"}):
+        raise ReadbackError("protected_configuration_unavailable")
+    require_identity({"RUNTIME_TARGET_JSON": env.get("EXPECTED_RUNTIME_TARGET_JSON"),
+                      "LONGBRIDGE_PHYSICAL_ACCOUNT_ID": account,
+                      "RUNTIME_TARGET_ENABLED": env.get("EXPECTED_RUNTIME_TARGET_ENABLED"),
+                      "LONGBRIDGE_DRY_RUN_ONLY": env.get("EXPECTED_DRY_RUN_ONLY")},
+                     service=service, account=account)
+    return service, region, project, account
+
+
+def observe(*, env=os.environ, load=read_json):
+    service, region, project, account = configuration(env)
+    started = datetime.now(UTC)
+    flags = ["--project=" + project, "--region=" + region, "--format=json"]
+    service_cmd = ["gcloud", "run", "services", "describe", service, *flags]
+    before = load(service_cmd)
+    if before.get("metadata", {}).get("name") != service:
+        raise ReadbackError("service_identity_invalid")
+    template = container_env(before.get("spec", {}).get("template", {}).get("spec", {}))
+    require_identity(template, service=service, account=account)
+    traffic = before.get("status", {}).get("traffic")
+    if not isinstance(traffic, list) or not traffic or len(traffic) > 100:
+        raise ReadbackError("traffic_unavailable")
+    revisions = {}
+    for row in traffic:
+        percent = row.get("percent", 0)
+        if type(percent) is not int or not 0 <= percent <= 100:
+            raise ReadbackError("traffic_unavailable")
+        if percent:
+            name = row.get("revisionName")
+            if (not isinstance(name, str) or not re.fullmatch(re.escape(service) + r"-[a-z0-9-]{1,63}", name)
+                    or name in revisions):
+                raise ReadbackError("revision_identity_invalid")
+            revisions[name] = percent
+    if sum(revisions.values()) != 100:
+        raise ReadbackError("traffic_unavailable")
+    values = [template]
+    for name in revisions:
+        cmd = ["gcloud", "run", "revisions", "describe", name, *flags]
+        revision = load(cmd)
+        if revision.get("metadata", {}).get("name") != name:
+            raise ReadbackError("revision_identity_invalid")
+        current = container_env(revision.get("spec", {}))
+        require_identity(current, service=service, account=account)
+        values.append(current)
+        if load(cmd) != revision:
+            raise ReadbackError("revision_changed_during_observation")
+    if load(service_cmd) != before:
+        raise ReadbackError("service_changed_during_observation")
+    return {"schema": "longbridge.plain-dca-paper-readback.v1", "status": "partial",
+            "observed_started_at": started.isoformat(),
+            "observed_finished_at": datetime.now(UTC).isoformat(),
+            "configured_paper_identity_verified": True, "runtime_target_enabled": False,
+            "physical_account_matches_protected": True,
+            "positive_traffic_revision_count": len(revisions), "service_stable": True,
+            "live_command_enabled": True if any(v.get("LONGBRIDGE_DURABLE_EXECUTION_COMMAND_LIVE_ENABLED") == "true"
+                                                 for v in values) else (
+                False if all(v.get("LONGBRIDGE_DURABLE_EXECUTION_COMMAND_LIVE_ENABLED") == "false"
+                             for v in values) else None),
+            "kelly_enabled": None, "stop_loss_enabled": None,
+            "reason": "authoritative_safety_source_unavailable", "paper_start_ready": False,
+            "broker_identity_independently_verified": False,
+            "service_invoked": False, "broker_invoked": False, "business_writes": False}
+
+
+def main(argv=None):
+    try:
+        if argv == ["--validate-config"]:
+            configuration(os.environ)
+            return 0
+        result = observe()
+    except Exception as exc:  # noqa: BLE001 - Never render protected cloud errors.
+        # No exception text, service identifiers, account values or cloud stderr.
+        result = {"schema": "longbridge.plain-dca-paper-readback.v1", "status": "unknown",
+                  "reason": "protected_readback_unavailable", "paper_start_ready": False}
+        if isinstance(exc, ReadbackError):
+            result["reason"] = str(exc)
+    print(json.dumps(result, sort_keys=True))
+    return 1  # A partial/unknown observation never grants readiness.
+
+
+if __name__ == "__main__":
+    import sys
+    raise SystemExit(main(sys.argv[1:]))

tests/test_plain_dca_paper_adapter.py

@@ -88,6 +88,28 @@ def test_exact_paper_binding_passes():
     assert check_plain_dca_paper_binding(readback(), now=NOW) is None
 
 
+@pytest.mark.parametrize("name", ["QSL_RUNTIME_TARGET_ENABLED", "QSL_LONGBRIDGE_DRY_RUN_ONLY"])
+@pytest.mark.parametrize("value", ["true", None, "", 0])
+def test_conflicting_or_unresolved_runtime_alias_blocks(name, value):
+    binding = readback()
+    binding["protected_env"][name] = value
+    with pytest.raises(PlainDcaPaperError):
+        check_plain_dca_paper_binding(binding, now=NOW)
+
+
+def test_identical_aliases_pass_and_conflicting_target_alias_blocks():
+    binding = readback()
+    env = binding["protected_env"]
+    env.update(QSL_RUNTIME_TARGET_ENABLED="false", QSL_LONGBRIDGE_DRY_RUN_ONLY="false",
+               QSL_RUNTIME_TARGET_JSON=env["RUNTIME_TARGET_JSON"])
+    assert check_plain_dca_paper_binding(binding, now=NOW) is None
+    target = json.loads(env["RUNTIME_TARGET_JSON"])
+    target["execution_environment"] = "live"
+    env["QSL_RUNTIME_TARGET_JSON"] = json.dumps(target)
+    with pytest.raises(PlainDcaPaperError):
+        check_plain_dca_paper_binding(binding, now=NOW)
+
+
 @pytest.mark.parametrize("field", ["live_enabled", "kelly_enabled", "stop_loss_enabled"])
 @pytest.mark.parametrize("state", [True, None, "false", 0])
 def test_unknown_or_enabled_safety_blocks(field, state):