feat: Add a paper-only plain-DCA preview consumer

application/plain_dca_paper_adapter.py

@@ -0,0 +1,115 @@
+"""Phase 1 plain-DCA consumer. It can return previews, never broker orders.
+
+The trusted caller supplies a fresh protected binding readback on every call.
+Labels are checked against the manifest and explicit broker PAPER configuration;
+this does not authenticate credentials or establish deployed account isolation.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from datetime import datetime, timedelta
+import importlib
+import json
+from typing import Any
+
+from application.runtime_target_manifest import load_runtime_target_manifest
+
+PAPER_ENVIRONMENT = "longbridge-paper"
+_BASELINES = frozenset({"nasdaq100_buy_and_hold", "sp500_buy_and_hold"})
+
+
+class PlainDcaPaperError(ValueError):
+    """Fixed diagnostic codes only; protected values must never be rendered."""
+
+
+def _paper_account_from_readback(readback: object, *, now: datetime) -> str:
+    if not isinstance(readback, Mapping):
+        raise PlainDcaPaperError("paper_binding_readback_invalid")
+    snapshot = dict(readback)
+    observed, expires = snapshot.get("observed_at"), snapshot.get("valid_until")
+    if (not all(isinstance(value, datetime) and value.utcoffset() is not None
+                for value in (now, observed, expires))
+            or not observed <= now < expires or now - observed > timedelta(minutes=5)):
+        raise PlainDcaPaperError("paper_binding_readback_stale")
+    if snapshot.get("environment") != PAPER_ENVIRONMENT or snapshot.get("target_id") != "paper":
+        raise PlainDcaPaperError("paper_target_environment_required")
+    if any(snapshot.get(key) is not False for key in (
+        "live_enabled", "kelly_enabled", "stop_loss_enabled",
+    )):
+        raise PlainDcaPaperError("paper_safety_off_not_proven")
+    protected = snapshot.get("protected_env")
+    if not isinstance(protected, Mapping):
+        raise PlainDcaPaperError("paper_protected_binding_missing")
+    env = dict(protected)
+    try:
+        manifest = load_runtime_target_manifest()
+        entries = [entry for entry in manifest.targets if entry.id == "paper"]
+        target = json.loads(env.get("RUNTIME_TARGET_JSON", ""))
+    except Exception:
+        raise PlainDcaPaperError("paper_target_readback_invalid") from None
+    if (len(entries) != 1 or entries[0].environment != PAPER_ENVIRONMENT
+            or entries[0].mode != "paper" or not isinstance(target, dict)):
+        raise PlainDcaPaperError("paper_target_readback_invalid")
+    if (env.get("CLOUD_RUN_SERVICE") != entries[0].service
+            or target.get("service_name") != entries[0].service
+            or target.get("platform_id") != "longbridge"
+            or target.get("execution_environment") != "paper"
+            or target.get("execution_mode") != "live"
+            or target.get("dry_run_only") is not False
+            or target.get("account_scope") != "PAPER"
+            or target.get("account_selector") != ["PAPER"]
+            or target.get("deployment_selector") != "PAPER"):
+        raise PlainDcaPaperError("paper_protected_target_mismatch")
+    # 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"):
+        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()
+            or account.upper() in {"PAPER", "LIVE", "SG", "HK", "US", "DEFAULT"}):
+        raise PlainDcaPaperError("paper_physical_account_binding_required")
+    return account
+
+
+def check_plain_dca_paper_binding(readback: object, *, now: datetime) -> None:
+    """Reject any unresolved, non-paper, enabled or unknown protected binding."""
+    _paper_account_from_readback(readback, now=now)
+
+
+def consume_plain_dca_preview(
+    *, read_binding: Callable[[], Mapping[str, object]], plan: Any, calendar: Any,
+    period: Any, facts: Any, quote: Any, capabilities: Any, now: Any,
+) -> Any:
+    """Consume UES's exact preview contract without scaling or persisting it.
+
+    No cached admission, period claim, simulated fill, schedule or broker call is
+    created. A future order writer still needs atomic reread/claim/fencing and
+    authentic isolation/qualification evidence. Reader failures are terminal for
+    this attempt. Restarting the caller does not retain an earlier admission.
+    """
+    try:
+        readback = read_binding()
+    except Exception:
+        raise PlainDcaPaperError("paper_binding_readback_unavailable") from None
+    account = _paper_account_from_readback(readback, now=now)
+    binding = getattr(getattr(plan, "mandate", None), "binding", None)
+    if (getattr(binding, "environment", None) != "paper"
+            or getattr(binding, "account_ref", None) != account
+            or getattr(binding, "profile", None) not in _BASELINES):
+        raise PlainDcaPaperError("plain_dca_plan_binding_mismatch")
+    try:
+        api = importlib.import_module("us_equity_strategies.plain_dca")
+        result = api.evaluate_dca_period(
+            plan=plan, calendar=calendar, period=period, facts=facts, quote=quote,
+            capabilities=capabilities, now=now,
+        )
+    except (ImportError, AttributeError):
+        raise PlainDcaPaperError("plain_dca_api_unavailable") from None
+    except Exception:
+        raise PlainDcaPaperError("plain_dca_evaluation_failed") from None
+    if (not isinstance(result, api.PeriodPreview)
+            or result.no_order is not True or result.no_execute is not True
+            or result.preview_only is not True or result.live_ready is not False):
+        raise PlainDcaPaperError("plain_dca_preview_contract_invalid")
+    return result

docs/runtime-target-manifest.md

@@ -173,3 +173,52 @@ configuration alongside its injected authorized session. Request-log and HK
 stop-result tools require `GCP_PROJECT_ID` in their process environment.
 `STRATEGY_PLUGIN_ALLOWED_SIGNAL_PREFIXES` supplies any additional approved
 plugin-signal prefixes; no project-specific bucket is a built-in fallback.
+
+## Plain-DCA Phase 1 consumer binding
+
+`application.plain_dca_paper_adapter.consume_plain_dca_preview` consumes the UES
+`evaluate_dca_period` preview interface for `nasdaq100_buy_and_hold` and
+`sp500_buy_and_hold`. It is an explicit library entry point; no route, scheduler,
+strategy registry, deployment workflow or order writer is attached to it.
+
+The trusted execution owner must provide `read_binding`, a callable that reads
+current protected configuration on every attempt. Its readback contains:
+
+- `target_id=paper` and `environment=longbridge-paper`, resolved by the protected
+  caller, with timezone-aware `observed_at` and `valid_until`. Evidence older
+  than five minutes, expired evidence and future observations are rejected.
+- Literal false `live_enabled`, `kelly_enabled` and `stop_loss_enabled` evidence.
+  The last field attests that **all** applicable stop-loss switches are OFF.
+  Missing or unknown runtime evidence must stay unknown, never be filled with
+  false merely to pass this check.
+- `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.
+
+`check_plain_dca_paper_binding(readback, now=now)` checks this contract. It rejects
+`longbridge-sg`, `longbridge-hk`, other target ids, service disagreement with the
+manifest, non-PAPER selectors and any target lacking explicit
+`execution_environment=paper`. The plan must bind the same physical account,
+PAPER environment and one of the two baseline profiles. No fallback to the
+legacy smart-DCA strategy is allowed. Protected values are never logged or
+included in exception messages.
+
+In the shared runtime contract, `dry_run_only=false` with legacy
+`execution_mode=live` is compatible with broker PAPER **only when** the explicit
+`execution_environment=paper` is present. False alone cannot establish broker
+PAPER identity. Phase 1 also requires `RUNTIME_TARGET_ENABLED=false`; the adapter
+cannot arm execution even with an admissible binding.
+
+Inputs are passed unchanged to UES, including approved final tranche, calendar,
+period records, quote, cash and capability evidence. There is no extra tier or
+Kelly scaling, catch-up, anchor acceleration or synthetic fill. All four preview
+flags remain enforced. The adapter neither claims periods nor reserves cash, so
+it makes no exactly-once or multi-process order-execution claim.
+
+The current locked UES package does not supply `plain_dca`. Until a separately
+reviewed compatible pin is adopted, a valid binding still fails with
+`plain_dca_api_unavailable`. This patch does not change shared dependency pins or
+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.

tests/test_plain_dca_paper_adapter.py

@@ -0,0 +1,196 @@
+"""Fake-only Phase 1 tests. The real UES contract is independently qualified."""
+from copy import deepcopy
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import importlib
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from application.plain_dca_paper_adapter import (
+    PlainDcaPaperError, check_plain_dca_paper_binding, consume_plain_dca_preview,
+)
+
+
+NOW = datetime(2026, 11, 5, tzinfo=timezone.utc)
+
+
+@dataclass(frozen=True)
+class Preview:
+    no_order: bool = True
+    no_execute: bool = True
+    preview_only: bool = True
+    live_ready: bool = False
+
+
+def readback():
+    return {
+        "observed_at": NOW, "valid_until": NOW + timedelta(minutes=1),
+        "target_id": "paper", "environment": "longbridge-paper",
+        "live_enabled": False, "kelly_enabled": False, "stop_loss_enabled": False,
+        "protected_env": {
+            "CLOUD_RUN_SERVICE": "longbridge-quant-paper-service",
+            "RUNTIME_TARGET_ENABLED": "false", "LONGBRIDGE_DRY_RUN_ONLY": "false",
+            "LONGBRIDGE_PHYSICAL_ACCOUNT_ID": "fake-paper-account",
+            "RUNTIME_TARGET_JSON": json.dumps({
+                "platform_id": "longbridge", "service_name": "longbridge-quant-paper-service",
+                "execution_environment": "paper", "dry_run_only": False,
+                "execution_mode": "live", "account_scope": "PAPER",
+                "account_selector": ["PAPER"], "deployment_selector": "PAPER",
+            }),
+        },
+    }
+
+
+def inputs():
+    binding = SimpleNamespace(environment="paper", account_ref="fake-paper-account",
+                              profile="nasdaq100_buy_and_hold")
+    return dict(plan=SimpleNamespace(mandate=SimpleNamespace(binding=binding)),
+                calendar=None, period=None, facts=None, quote=None, capabilities=None,
+                now=NOW)
+
+
+@pytest.fixture
+def planner(monkeypatch):
+    calls = []
+
+    def evaluate(**kwargs):
+        calls.append(kwargs)
+        return Preview()
+
+    api = SimpleNamespace(evaluate_dca_period=evaluate, PeriodPreview=Preview)
+    original = importlib.import_module
+    monkeypatch.setattr(importlib, "import_module", lambda name: api if name ==
+                        "us_equity_strategies.plain_dca" else original(name))
+    return calls, api
+
+
+@pytest.mark.parametrize("environment", ["longbridge-sg", "longbridge-hk", "", None,
+                                         "longbridge-paper ", "LONGBRIDGE-PAPER"])
+def test_reject_non_paper_environment_before_planner(environment, planner):
+    value = readback()
+    value["environment"] = environment
+    with pytest.raises(PlainDcaPaperError):
+        consume_plain_dca_preview(read_binding=lambda: value, **inputs())
+    assert planner[0] == []
+
+
+@pytest.mark.parametrize("target", ["sg", "hk", "longbridge-sg", "", None])
+def test_reject_other_resolved_targets(target):
+    value = readback()
+    value["target_id"] = target
+    with pytest.raises(PlainDcaPaperError):
+        check_plain_dca_paper_binding(value, now=NOW)
+
+
+def test_exact_paper_binding_passes():
+    assert check_plain_dca_paper_binding(readback(), now=NOW) is None
+
+
+@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):
+    value = readback()
+    value[field] = state
+    with pytest.raises(PlainDcaPaperError):
+        check_plain_dca_paper_binding(value, now=NOW)
+
+
+@pytest.mark.parametrize("patch", [
+    {"execution_environment": "live"}, {"execution_environment": None},
+    {"account_scope": "SG"}, {"account_selector": ["PAPER", "SG"]},
+    {"deployment_selector": "SG"}, {"platform_id": "fake-other"},
+    {"service_name": "fake-other-service"}, {"dry_run_only": True},
+])
+def test_protected_target_mismatch_blocks(patch):
+    value = readback()
+    target = json.loads(value["protected_env"]["RUNTIME_TARGET_JSON"])
+    target.update(patch)
+    value["protected_env"]["RUNTIME_TARGET_JSON"] = json.dumps(target)
+    with pytest.raises(PlainDcaPaperError):
+        check_plain_dca_paper_binding(value, now=NOW)
+
+
+@pytest.mark.parametrize("field,value", [
+    ("RUNTIME_TARGET_JSON", "{fake-private-value"),
+    ("RUNTIME_TARGET_ENABLED", "true"), ("RUNTIME_TARGET_ENABLED", ""),
+    ("LONGBRIDGE_DRY_RUN_ONLY", ""), ("CLOUD_RUN_SERVICE", "fake-other-service"),
+    ("LONGBRIDGE_PHYSICAL_ACCOUNT_ID", "PAPER"),
+])
+def test_invalid_protected_values_fail_without_echo(field, value, capsys):
+    snapshot = readback()
+    snapshot["protected_env"][field] = value
+    with pytest.raises(PlainDcaPaperError) as exc:
+        check_plain_dca_paper_binding(snapshot, now=NOW)
+    assert "fake-private-value" not in str(exc.value)
+    assert capsys.readouterr() == ("", "")
+
+
+def test_consumer_passes_inputs_and_preserves_preview(planner):
+    args = inputs()
+    assert consume_plain_dca_preview(read_binding=readback, **args) == Preview()
+    assert planner[0] == [args]
+
+
+def test_each_call_rereads_binding_across_target_change_and_restart(planner):
+    value = readback()
+
+    def reader():
+        return deepcopy(value)
+
+    consume_plain_dca_preview(read_binding=reader, **inputs())
+    value["environment"] = "longbridge-sg"
+    for current_reader in (reader, lambda: deepcopy(value)):
+        with pytest.raises(PlainDcaPaperError):
+            consume_plain_dca_preview(read_binding=current_reader, **inputs())
+    assert len(planner[0]) == 1
+
+
+def test_reader_failure_is_static_and_never_calls_planner(planner, capsys):
+    def unavailable():
+        raise OSError("fake-private-value")
+
+    with pytest.raises(PlainDcaPaperError) as exc:
+        consume_plain_dca_preview(read_binding=unavailable, **inputs())
+    assert str(exc.value) == "paper_binding_readback_unavailable"
+    assert planner[0] == []
+    assert capsys.readouterr() == ("", "")
+
+
+def test_plan_cannot_use_another_account_or_offline_binding(planner):
+    for field, value in (("account_ref", "fake-other-account"),
+                         ("environment", "offline_preview"), ("profile", "fake-other")):
+        args = inputs()
+        setattr(args["plan"].mandate.binding, field, value)
+        with pytest.raises(PlainDcaPaperError):
+            consume_plain_dca_preview(read_binding=readback, **args)
+    assert planner[0] == []
+
+
+def test_missing_installed_ues_api_blocks(monkeypatch):
+    def missing(name):
+        raise ModuleNotFoundError("fake-private-value")
+
+    monkeypatch.setattr(importlib, "import_module", missing)
+    with pytest.raises(PlainDcaPaperError, match="^plain_dca_api_unavailable$"):
+        consume_plain_dca_preview(read_binding=readback, **inputs())
+
+
+def test_unsafe_planner_result_is_not_forwarded(planner):
+    planner[1].evaluate_dca_period = lambda **kwargs: Preview(no_execute=False)
+    with pytest.raises(PlainDcaPaperError, match="^plain_dca_preview_contract_invalid$"):
+        consume_plain_dca_preview(read_binding=readback, **inputs())
+
+
+@pytest.mark.parametrize("patch", [
+    {"observed_at": NOW - timedelta(minutes=6)},
+    {"observed_at": NOW + timedelta(seconds=1)}, {"valid_until": NOW},
+    {"observed_at": None}, {"observed_at": NOW.replace(tzinfo=None)},
+])
+def test_stale_future_or_unknown_readback_blocks(patch, planner):
+    value = readback()
+    value.update(patch)
+    with pytest.raises(PlainDcaPaperError, match="^paper_binding_readback_stale$"):
+        consume_plain_dca_preview(read_binding=lambda: value, **inputs())
+    assert planner[0] == []