From 72755351ff01a009eff4e697530b000ac42c57fb Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 22:28:42 +0800 Subject: [PATCH 1/4] fix: persist account drawdown blocks for new-risk submissions --- docs/account-new-risk-gate.zh-CN.md | 21 +++ .../common/execution_state.py | 49 ++++++- tests/test_persistent_drawdown_breaker.py | 136 ++++++++++++++++++ 3 files changed, 202 insertions(+), 4 deletions(-) create mode 100644 tests/test_persistent_drawdown_breaker.py diff --git a/docs/account-new-risk-gate.zh-CN.md b/docs/account-new-risk-gate.zh-CN.md index 42df3bad..a113f71d 100644 --- a/docs/account-new-risk-gate.zh-CN.md +++ b/docs/account-new-risk-gate.zh-CN.md @@ -118,3 +118,24 @@ W1 平台 adapter 已有实现；持久化 OPEN 熔断、实际账户读回和 `target_vol=None` 使用既有默认值；显式非正数、非有限数、布尔值或非数值会禁止新增风险。非法 realized volatility、drawdown 或 equity 同样禁止新增风险。信封数值缩放仅用于诊断，执行方必须检查 `new_risk_allowed` 和原因。 `NewRiskAdmissionResult` 新增默认 false 的 `risk_input_invalid` 和 `hard_brake_triggered`。数据错误且没有可信 drawdown 硬刹车时，`combined_scale=None` 不表示零仓位；可信硬刹车同时存在时，两个标记均为 true，缩放保留 0。`apply_envelope_to_sized_weight` 对数据错误返回 `None`，可信硬刹车同时存在时返回 0。调用方必须处理可选返回值；升级前应完成消费者动作联测。 + +## 账户日志中的持久化回撤 OPEN + +`AccountSubmissionSession.record_circuit_breaker_open(snapshot=...)` 接受现有回撤门确认的 +硬刹车事实，在同一物理账户的 `account_submission.v1` 追加日志中写入 +`CIRCUIT_BREAKER_OPEN`。沿用排他创建、连续序号、哈希链及完整读回；不新建状态文件。 +事件只含注入的 equity/peak/drawdown 与调用方身份，不证明其真实来源或资金流口径。 + +`read_state()` 增加 `circuit_breaker_open` 布尔字段。合法旧日志初始为 false；已记录 +OPEN 则持续为 true，包括其他 profile 及进程重启。恢复观测、CLOSED、reset 形状事件 +都不能清除它；本次不提供复位 API。无效输入、冲突、损坏或未确认写入须禁止新增风险。 + +平台使用 `reserve(order, new_risk=True)` 将新增风险类别写入 PREPARED；OPEN 时拒绝 +reserve，dispatch 也重新检查 OPEN，阻止已准备但尚未发送的新增风险。省略该参数的 +旧调用保持原语义，仍须由各自平台完成类别判断和门控；本次接线范围为 IBKR 的买单。 +OPEN 不创建目标、不产生卖单、不替代既有减仓证明，不改变 UNKNOWN 的含义。 + +这是追加事件类型：旧版本读取新事件会拒绝，不能把混用旧写入器视为安全迁移。 +消费者升级须先审核库和平台补丁，再单独协调固定版本、旧写入器停止与运行验证； +本次不修改任何依赖固定版本或部署配置。离线合成日志不能证明真实初始状态、旧写入器 +已停止、实际消费者已升级或实盘行为；这些验收保持 NOT_PROVEN。 diff --git a/src/quant_platform_kit/common/execution_state.py b/src/quant_platform_kit/common/execution_state.py index acf74f07..034f2a0f 100644 --- a/src/quant_platform_kit/common/execution_state.py +++ b/src/quant_platform_kit/common/execution_state.py @@ -925,7 +925,7 @@ def read(path): return path.read_text(encoding="utf-8") if len(locations) != len(set(map(str, locations))): raise AccountSubmissionError("duplicate account journal sequence") - state: dict[str, Any] = {"generation": 0, "head_sha256": "0" * 64, "attempts": {}, "domain_binding": None, "risk_reservations": {}, "equity_observation": None, "snapshot_position_risk": 0.0} + state: dict[str, Any] = {"generation": 0, "head_sha256": "0" * 64, "attempts": {}, "domain_binding": None, "risk_reservations": {}, "equity_observation": None, "snapshot_position_risk": 0.0, "circuit_breaker_open": False} for location in sorted(locations, key=str): sequence = state["generation"] + 1 if str(location) != str(self._location(sequence)): @@ -945,7 +945,21 @@ def read(path): data = event["data"] if not isinstance(data, dict): raise AccountSubmissionError("invalid account event data") - if event["kind"] == "DOMAIN_BOUND": + if event["kind"] == "CIRCUIT_BREAKER_OPEN": + from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope + if set(data) != {"equity_usd", "peak_equity_usd", "drawdown_from_peak", "owner_id", "profile"} or state["circuit_breaker_open"]: + raise AccountSubmissionError("invalid or duplicate circuit breaker OPEN") + _submission_text(data["owner_id"]) + _submission_text(data["profile"]) + equity = _submission_number(data["equity_usd"]) + peak = None if data["peak_equity_usd"] is None else _submission_number(data["peak_equity_usd"]) + drawdown = None if data["drawdown_from_peak"] is None else _submission_number(data["drawdown_from_peak"]) + if drawdown is None and peak is not None and peak > 0: + drawdown = max(0.0, 1.0 - equity / peak) + if "DRAWDOWN_BRAKE_TRIPPED" not in evaluate_capital_risk_envelope(equity, drawdown_from_peak=drawdown).reasons: + raise AccountSubmissionError("circuit breaker OPEN requires a drawdown breach") + state["circuit_breaker_open"] = True + elif event["kind"] == "DOMAIN_BOUND": if set(data) != {"domain_key", "account_type", "currency"} or state["domain_binding"] is not None: raise AccountSubmissionError("invalid or duplicate domain binding") _submission_digest(data["domain_key"]) @@ -1048,8 +1062,10 @@ def read(path): state["snapshot_position_risk"] = position_risk elif event["kind"] == "PREPARED": keys = {"attempt_id", "fence", "intent_id", "order_sha256", "release_id", "owner_id", "profile", "quantity"} - if set(data) not in (keys, keys | {"reduction", "reduction_source_digest"}) or ("reduction" in data and data["reduction"] is not True): + if set(data) not in (keys, keys | {"reduction", "reduction_source_digest"}, keys | {"new_risk"}) or ("reduction" in data and data["reduction"] is not True): raise AccountSubmissionError("invalid prepared fields") + if "new_risk" in data and (data["new_risk"] is not True or state["circuit_breaker_open"]): + raise AccountSubmissionError("new risk prepared while circuit breaker OPEN") for key in ["attempt_id", "fence", "intent_id", "order_sha256", "release_id"]: _submission_digest(data[key]) for key in ["owner_id", "profile"]: @@ -1146,6 +1162,25 @@ def _append(self, state: Mapping[str, Any], kind: str, data: dict) -> dict: raise return event + def record_circuit_breaker_open(self, *, snapshot): + """Latch an observed drawdown breach in the existing account journal. + + OPEN is admission-only. Recovery, timeouts and caller CLOSED/reset flags + cannot clear it. This method grants no reduction or submission authority. + """ + from quant_platform_kit.risk.account_new_risk_gate import evaluate_new_risk_admission + with self._mutex: + if self._failed: + raise AccountSubmissionError("session persistence uncertain") + if not evaluate_new_risk_admission(snapshot).hard_brake_triggered: + raise AccountSubmissionError("circuit breaker OPEN requires a drawdown breach") + data = {key: None if getattr(snapshot, key) is None else _submission_number(getattr(snapshot, key)) + for key in ("equity_usd", "peak_equity_usd", "drawdown_from_peak")} + data.update(owner_id=self.owner_id, profile=self.profile) + state = self.read_state() + if not state["circuit_breaker_open"]: + self._append(state, "CIRCUIT_BREAKER_OPEN", data) + def record_account_equity(self, *, snapshot, cash_flow, baseline_confirmed=False): """Record a complete account equity observation with cash-flow-adjusted peak.""" from quant_platform_kit.common.account_risk_budget import available_new_risk_budget @@ -1430,13 +1465,15 @@ def reserve_reduction(self, order, *, admission) -> AccountSubmissionGrant: self._issued[grant.attempt_id] = grant return grant - def reserve(self, order) -> AccountSubmissionGrant: + def reserve(self, order, *, new_risk=False) -> AccountSubmissionGrant: """Only a fresh acknowledged create can return a grant; never reclaim it.""" with self._mutex: if self._failed: raise AccountSubmissionError("session persistence uncertain; no grant") digest, quantity = self._order(order) state = self.read_state() + if type(new_risk) is not bool or (new_risk and state["circuit_breaker_open"]): + raise AccountSubmissionError("new risk prohibited by circuit breaker") if self.account_domain is not None: from quant_platform_kit.common.account_domain import ( account_domain_key, account_domain_matches_broker_identity, account_domain_policy_active, validate_account_domain, @@ -1481,6 +1518,8 @@ def reserve(self, order) -> AccountSubmissionGrant: "intent_id": self.intent_id, "order_sha256": digest, "release_id": self.release_id, "owner_id": self.owner_id, "profile": self.profile, "quantity": quantity, } + if new_risk: + data["new_risk"] = True event = self._append(state, "PREPARED", data) grant = AccountSubmissionGrant(self.account_root, data["attempt_id"], data["fence"], event["sequence"], event["event_sha256"], self.intent_id, digest, self.release_id) @@ -1512,6 +1551,8 @@ def dispatch(self, grant: AccountSubmissionGrant, order, submit: Callable[[], An raise AccountSubmissionError("wrong order or already-used grant") if state["attempts"][grant.attempt_id]["certainty"] != "PREPARED": raise AccountSubmissionError("grant no longer prepared") + if state["attempts"][grant.attempt_id].get("new_risk") and state["circuit_breaker_open"]: + raise AccountSubmissionError("new risk prohibited by circuit breaker") if self.account_domain is not None: from quant_platform_kit.common.account_domain import ( _time, account_domain_key, account_domain_matches_broker_identity, diff --git a/tests/test_persistent_drawdown_breaker.py b/tests/test_persistent_drawdown_breaker.py new file mode 100644 index 00000000..476b71e5 --- /dev/null +++ b/tests/test_persistent_drawdown_breaker.py @@ -0,0 +1,136 @@ +"""Synthetic durable admission-only drawdown breaker regressions.""" +from dataclasses import replace +from types import SimpleNamespace + +import pytest + +from quant_platform_kit.common.execution_state import ( + AccountSubmissionError, AccountSubmissionSession, ExecutionMarkerStore, +) +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot + + +def session_at(path, *, account="DU00000001", profile="synthetic"): + return AccountSubmissionSession( + ExecutionMarkerStore(local_dir=path, cloud_prefix_uri=None), + broker="ibkr", account_id=account, owner_id=profile, profile=profile, + intent_id="1" * 64, release_id="2" * 64, + ) + + +def breached_snapshot(): + return InjectedReconciliationSnapshot( + observation_status="COMPLETE", reconciliation_status="VERIFIED", + circuit_breaker_state="CLOSED", equity_usd=40_000, + peak_equity_usd=50_000, drawdown_from_peak=.2, + ) + + +def order(): + return SimpleNamespace(account_id="DU00000001", symbol="TQQQ", side="buy", quantity=1) + + +def test_open_survives_new_session_and_other_profile_without_reset(tmp_path): + first = session_at(tmp_path) + first.record_circuit_breaker_open(snapshot=breached_snapshot()) + second = session_at(tmp_path, profile="other-synthetic") + assert second.read_state()["circuit_breaker_open"] is True + second.record_circuit_breaker_open(snapshot=breached_snapshot()) + assert second.read_state()["generation"] == 1 + assert session_at(tmp_path, account="DU00000002").read_state()["circuit_breaker_open"] is False + with pytest.raises(AccountSubmissionError, match="breaker"): + second.reserve(order(), new_risk=True) + + +def test_prepared_new_risk_cannot_dispatch_after_another_session_trips(tmp_path): + first = session_at(tmp_path) + grant = first.reserve(order(), new_risk=True) + session_at(tmp_path, profile="other-synthetic").record_circuit_breaker_open(snapshot=breached_snapshot()) + calls = [] + with pytest.raises(AccountSubmissionError, match="breaker"): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert first.read_state()["attempts"][grant.attempt_id]["certainty"] == "PREPARED" + + +@pytest.mark.parametrize("snapshot", [ + replace(breached_snapshot(), drawdown_from_peak=0, peak_equity_usd=40_000), + replace(breached_snapshot(), equity_usd=float("nan")), + replace(breached_snapshot(), drawdown_from_peak=True, peak_equity_usd=None), +]) +def test_nonbreach_or_invalid_input_cannot_create_open(tmp_path, snapshot): + session = session_at(tmp_path) + with pytest.raises((AccountSubmissionError, ValueError)): + session.record_circuit_breaker_open(snapshot=snapshot) + assert session.read_state()["generation"] == 0 + + +def test_corrupt_and_reset_shaped_events_fail_closed(tmp_path): + session = session_at(tmp_path) + session.record_circuit_breaker_open(snapshot=breached_snapshot()) + path = session._location(1) + original = path.read_text() + import json + event = json.loads(original) + event["kind"] = "CIRCUIT_BREAKER_RESET" + event["event_sha256"] = session._event_hash(event) + path.write_text(json.dumps(event)) + with pytest.raises(AccountSubmissionError): + session.read_state() + path.write_text(original.replace('"sequence": 1', '"sequence": 2')) + with pytest.raises(AccountSubmissionError): + session.read_state() + + +def test_failed_write_and_readback_never_allow_new_risk(tmp_path, monkeypatch): + session = session_at(tmp_path) + monkeypatch.setattr("quant_platform_kit.common.execution_state.os.open", + lambda *_a, **_k: (_ for _ in ()).throw(OSError("synthetic write failure"))) + with pytest.raises(OSError): + session.record_circuit_breaker_open(snapshot=breached_snapshot()) + assert session._failed + with pytest.raises(AccountSubmissionError): + session.reserve(order(), new_risk=True) + + +def test_restart_in_separate_process_restores_open(tmp_path): + import subprocess + import sys + from pathlib import Path + session_at(tmp_path).record_circuit_breaker_open(snapshot=breached_snapshot()) + code = ''' +import sys +sys.path.insert(0, sys.argv[1]) +from quant_platform_kit.common.execution_state import AccountSubmissionSession, ExecutionMarkerStore +s = AccountSubmissionSession(ExecutionMarkerStore(local_dir=sys.argv[2]), broker="ibkr", account_id="DU00000001", owner_id="other", profile="other", intent_id="3"*64, release_id="4"*64) +assert s.read_state()["circuit_breaker_open"] is True +''' + result = subprocess.run([sys.executable, "-c", code, str(Path(__file__).resolve().parents[1] / "src"), str(tmp_path)], capture_output=True, text=True) + assert result.returncode == 0, result.stderr + + +def test_conflicting_open_create_and_failed_confirmation_poison_session(tmp_path, monkeypatch): + session = session_at(tmp_path) + stale_state = session.read_state() + session_at(tmp_path, profile="other").record_circuit_breaker_open(snapshot=breached_snapshot()) + monkeypatch.setattr(AccountSubmissionSession, "read_state", lambda self: stale_state) + with pytest.raises(AccountSubmissionError, match="conflict"): + session.record_circuit_breaker_open(snapshot=breached_snapshot()) + assert session._failed + with pytest.raises(AccountSubmissionError): + session.reserve(order(), new_risk=True) + + +def test_failed_readback_poison_session(tmp_path, monkeypatch): + session = session_at(tmp_path) + original = AccountSubmissionSession.read_state + reads = [] + def read(self): + reads.append(True) + if len(reads) > 1: + raise OSError("synthetic readback failure") + return original(self) + monkeypatch.setattr(AccountSubmissionSession, "read_state", read) + with pytest.raises(OSError, match="readback"): + session.record_circuit_breaker_open(snapshot=breached_snapshot()) + assert session._failed From 1c2d9d7ee3b8358c55de927248afa4502beb3fd0 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 23:07:39 +0800 Subject: [PATCH 2/4] fix: preserve peak breaches and serialize final breaker checks --- docs/account-new-risk-gate.zh-CN.md | 7 + .../common/execution_state.py | 92 ++++++--- tests/test_persistent_drawdown_breaker.py | 184 +++++++++++++++++- 3 files changed, 250 insertions(+), 33 deletions(-) diff --git a/docs/account-new-risk-gate.zh-CN.md b/docs/account-new-risk-gate.zh-CN.md index a113f71d..160115f2 100644 --- a/docs/account-new-risk-gate.zh-CN.md +++ b/docs/account-new-risk-gate.zh-CN.md @@ -125,6 +125,9 @@ W1 平台 adapter 已有实现；持久化 OPEN 熔断、实际账户读回和 硬刹车事实，在同一物理账户的 `account_submission.v1` 追加日志中写入 `CIRCUIT_BREAKER_OPEN`。沿用排他创建、连续序号、哈希链及完整读回；不新建状态文件。 事件只含注入的 equity/peak/drawdown 与调用方身份，不证明其真实来源或资金流口径。 +显式 drawdown 无效但有效 equity/peak 已独立确认越界时，日志保留 equity/peak， +将 drawdown 记为 null，由读回重新计算回撤并验证 OPEN；无效值不能使该次越界丢失。 +原始观测的数据错误仍由风险门报告，不把无效值当作有效事实。 `read_state()` 增加 `circuit_breaker_open` 布尔字段。合法旧日志初始为 false；已记录 OPEN 则持续为 true，包括其他 profile 及进程重启。恢复观测、CLOSED、reset 形状事件 @@ -134,6 +137,10 @@ OPEN 则持续为 true，包括其他 profile 及进程重启。恢复观测、C reserve，dispatch 也重新检查 OPEN，阻止已准备但尚未发送的新增风险。省略该参数的 旧调用保持原语义，仍须由各自平台完成类别判断和门控；本次接线范围为 IBKR 的买单。 OPEN 不创建目标、不产生卖单、不替代既有减仓证明，不改变 UNKNOWN 的含义。 +dispatch 完成订单及账户验证后，在同一账户日志的进程内共享锁中重新读回 OPEN， +并持锁调用最终 submit；其他 session 的 OPEN 写入使用同一把锁。锁按日志位置区分， +不随 owner/profile 改变。本锁不提供跨进程或跨主机的分布式互斥；日志排他创建及 +既有账户 owner fence 仍须满足，不能据此声称多个实际写入器已安全协调。 这是追加事件类型：旧版本读取新事件会拒绝，不能把混用旧写入器视为安全迁移。 消费者升级须先审核库和平台补丁，再单独协调固定版本、旧写入器停止与运行验证； diff --git a/src/quant_platform_kit/common/execution_state.py b/src/quant_platform_kit/common/execution_state.py index 034f2a0f..ab1dd841 100644 --- a/src/quant_platform_kit/common/execution_state.py +++ b/src/quant_platform_kit/common/execution_state.py @@ -759,6 +759,8 @@ def _date_key(value: object) -> str: _SUBMISSION_ACK = frozenset({"PendingSubmit", "ApiPending", "ApiPendingSubmit", "Submitted", "PreSubmitted"}) _SUBMISSION_PARTIAL = frozenset({"Partial", "PartiallyFilled"}) _SUBMISSION_TERMINAL = frozenset({"Filled", "Rejected", "rejected", "Cancelled", "ApiCancelled"}) +_ACCOUNT_SUBMISSION_LOCKS: dict[str, Any] = {} +_ACCOUNT_SUBMISSION_LOCKS_MUTEX = threading.Lock() class AccountSubmissionError(RuntimeError): @@ -860,6 +862,11 @@ def __init__(self, store: ExecutionMarkerStore, *, broker: str, account_id: str, } for name, value in values.items(): object.__setattr__(self, name, value) + location = self._location() + lock_key = str(location) if store.cloud_prefix_uri else str(location.resolve()) + with _ACCOUNT_SUBMISSION_LOCKS_MUTEX: + account_mutex = _ACCOUNT_SUBMISSION_LOCKS.setdefault(lock_key, threading.RLock()) + object.__setattr__(self, "_account_mutex", account_mutex) def _location(self, sequence: int | None = None): suffix = self.account_root + "/" + (f"{sequence:020d}.json" if sequence is not None else "") @@ -1169,13 +1176,22 @@ def record_circuit_breaker_open(self, *, snapshot): cannot clear it. This method grants no reduction or submission authority. """ from quant_platform_kit.risk.account_new_risk_gate import evaluate_new_risk_admission - with self._mutex: + from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope + with self._mutex, self._account_mutex: if self._failed: raise AccountSubmissionError("session persistence uncertain") if not evaluate_new_risk_admission(snapshot).hard_brake_triggered: raise AccountSubmissionError("circuit breaker OPEN requires a drawdown breach") data = {key: None if getattr(snapshot, key) is None else _submission_number(getattr(snapshot, key)) - for key in ("equity_usd", "peak_equity_usd", "drawdown_from_peak")} + for key in ("equity_usd", "peak_equity_usd")} + drawdown = snapshot.drawdown_from_peak + if "INVALID_DRAWDOWN_FAIL_CLOSED" in evaluate_capital_risk_envelope( + snapshot.equity_usd, drawdown_from_peak=drawdown, + ).reasons: + # Admission already verified the independent equity/peak breach. + # Store only those valid facts; replay derives their drawdown. + drawdown = None + data["drawdown_from_peak"] = None if drawdown is None else _submission_number(drawdown) data.update(owner_id=self.owner_id, profile=self.profile) state = self.read_state() if not state["circuit_breaker_open"]: @@ -1575,36 +1591,48 @@ def dispatch(self, grant: AccountSubmissionGrant, order, submit: Callable[[], An or observed["source_digest"] != state["attempts"][grant.attempt_id]["reduction_source_digest"])) or not _time(observed["valuation_at"]) <= _time(now_text) < _time(observed["valid_until"])): raise AccountSubmissionError("expired or mismatched domain observation") - self._used.add(grant.attempt_id) - try: - report = submit() - except Exception as exc: - try: - self.record_observation(grant, None) - except Exception as persist_exc: - exc.add_note("account outcome persistence also failed: " + type(persist_exc).__name__) - object.__setattr__(self, "_failed", True) - raise - if record_observation_watermark: - try: - now = self.clock() - recorded_at = now.isoformat() if isinstance(now, datetime) else now - from quant_platform_kit.common.account_domain import _time - _time(recorded_at) - except Exception: - recorded_at = None - if recorded_at is not None: - self.record_observation(grant, report, recorded_at=recorded_at) - else: - self.record_observation(grant, report) - else: - self.record_observation(grant, report) - attempt = self.read_state()["attempts"][grant.attempt_id] - if attempt["certainty"] == "UNKNOWN" and str(getattr(report, "status", "")).strip() in ( - _SUBMISSION_ACK | _SUBMISSION_PARTIAL | _SUBMISSION_TERMINAL - ): - raise AccountSubmissionError("broker result facts are uncertain; original attempt retained") - return report + with self._account_mutex: + # Validation above can call platform code. Re-read after it and + # serialize the final check/submit with OPEN from other sessions + # on this account journal in the current process. + current = self._owned_grant(grant) + attempt = current["attempts"][grant.attempt_id] + if attempt["certainty"] != "PREPARED": + raise AccountSubmissionError("grant no longer prepared") + if attempt.get("new_risk") and current["circuit_breaker_open"]: + raise AccountSubmissionError("new risk prohibited by circuit breaker") + self._used.add(grant.attempt_id) + try: + report = submit() + except Exception as exc: + try: + self.record_observation(grant, None) + except Exception as persist_exc: + exc.add_note("account outcome persistence also failed: " + type(persist_exc).__name__) + object.__setattr__(self, "_failed", True) + raise + # Keep the outcome append in the same critical section so an + # OPEN writer cannot race its next sequence with OBSERVED. + if record_observation_watermark: + try: + now = self.clock() + recorded_at = now.isoformat() if isinstance(now, datetime) else now + from quant_platform_kit.common.account_domain import _time + _time(recorded_at) + except Exception: + recorded_at = None + if recorded_at is not None: + self.record_observation(grant, report, recorded_at=recorded_at) + else: + self.record_observation(grant, report) + else: + self.record_observation(grant, report) + attempt = self.read_state()["attempts"][grant.attempt_id] + if attempt["certainty"] == "UNKNOWN" and str(getattr(report, "status", "")).strip() in ( + _SUBMISSION_ACK | _SUBMISSION_PARTIAL | _SUBMISSION_TERMINAL + ): + raise AccountSubmissionError("broker result facts are uncertain; original attempt retained") + return report def record_observation(self, grant: AccountSubmissionGrant, report, *, recorded_at=None) -> bool: """Persist original facts. Exact replay is idempotent, never a new grant.""" diff --git a/tests/test_persistent_drawdown_breaker.py b/tests/test_persistent_drawdown_breaker.py index 476b71e5..adb1ba47 100644 --- a/tests/test_persistent_drawdown_breaker.py +++ b/tests/test_persistent_drawdown_breaker.py @@ -1,5 +1,10 @@ """Synthetic durable admission-only drawdown breaker regressions.""" from dataclasses import replace +import json +from pathlib import Path +import subprocess +import sys +import threading from types import SimpleNamespace import pytest @@ -7,7 +12,9 @@ from quant_platform_kit.common.execution_state import ( AccountSubmissionError, AccountSubmissionSession, ExecutionMarkerStore, ) -from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot +from quant_platform_kit.risk.account_new_risk_gate import ( + InjectedReconciliationSnapshot, evaluate_new_risk_admission, +) def session_at(path, *, account="DU00000001", profile="synthetic"): @@ -134,3 +141,178 @@ def read(self): with pytest.raises(OSError, match="readback"): session.record_circuit_breaker_open(snapshot=breached_snapshot()) assert session._failed + + +@pytest.mark.parametrize("invalid_drawdown", [ + float("nan"), float("inf"), float("-inf"), -1., True, False, + "unknown", "0.2", [], {}, +]) +def test_valid_peak_breach_with_invalid_drawdown_survives_process_restart(tmp_path, invalid_drawdown): + snapshot = replace(breached_snapshot(), drawdown_from_peak=invalid_drawdown) + admission = evaluate_new_risk_admission(snapshot) + assert admission.risk_input_invalid is True + assert admission.hard_brake_triggered is True + first = session_at(tmp_path) + first.record_circuit_breaker_open(snapshot=snapshot) + event = json.loads(first._location(1).read_text()) + assert event["kind"] == "CIRCUIT_BREAKER_OPEN" + assert event["data"]["equity_usd"] == 40_000 + assert event["data"]["peak_equity_usd"] == 50_000 + # Replay derives the breach from valid peak/equity; no invalid fact is stored. + assert event["data"]["drawdown_from_peak"] is None + code = ''' +import sys +sys.path.insert(0, sys.argv[1]) +from types import SimpleNamespace +from quant_platform_kit.common.execution_state import AccountSubmissionSession, AccountSubmissionError, ExecutionMarkerStore +s = AccountSubmissionSession(ExecutionMarkerStore(local_dir=sys.argv[2]), broker="ibkr", account_id="DU00000001", owner_id="other", profile="other", intent_id="3"*64, release_id="4"*64) +assert s.read_state()["circuit_breaker_open"] is True +try: + s.reserve(SimpleNamespace(account_id="DU00000001", symbol="SOXL", side="buy", quantity=1), new_risk=True) +except AccountSubmissionError: + pass +else: + raise AssertionError("restarted session admitted new risk after peak breach") +''' + result = subprocess.run([sys.executable, "-c", code, str(Path(__file__).resolve().parents[1] / "src"), str(tmp_path)], capture_output=True, text=True) + assert result.returncode == 0, result.stderr + recovered = replace(breached_snapshot(), equity_usd=50_000, drawdown_from_peak=0) + assert evaluate_new_risk_admission(recovered).hard_brake_triggered is False + restarted = session_at(tmp_path, profile="other") + with pytest.raises(AccountSubmissionError, match="breaker"): + restarted.reserve(order(), new_risk=True) + assert restarted.read_state()["generation"] == 1 + + +def test_open_completed_during_order_validation_blocks_final_dispatch(tmp_path, monkeypatch): + first = session_at(tmp_path) + grant = first.reserve(order(), new_risk=True) + validating, opened = threading.Event(), threading.Event() + failures, calls = [], [] + original = AccountSubmissionSession._order + + def validate(self, intent): + if self is first: + validating.set() + assert opened.wait(5) + return original(self, intent) + + def trip(): + try: + assert validating.wait(5) + second = session_at(tmp_path, profile="other") + second.record_circuit_breaker_open(snapshot=breached_snapshot()) + assert second.read_state()["circuit_breaker_open"] is True + except BaseException as exc: + failures.append(exc) + finally: + opened.set() + + monkeypatch.setattr(AccountSubmissionSession, "_order", validate) + worker = threading.Thread(target=trip) + worker.start() + try: + with pytest.raises(AccountSubmissionError, match="breaker"): + first.dispatch(grant, order(), lambda: calls.append(True)) + finally: + worker.join(5) + assert not worker.is_alive() + assert failures == [] + assert calls == [] + assert grant.attempt_id not in first._used + assert first.read_state()["attempts"][grant.attempt_id]["certainty"] == "PREPARED" + + +@pytest.mark.parametrize("cloud", [False, True]) +def test_open_and_final_read_submit_are_serialized_across_sessions(tmp_path, monkeypatch, cloud): + if cloud: + from tests.test_execution_state import _JournalObjects, _JournalStore + objects = _JournalObjects() + + def make_session(profile): + return AccountSubmissionSession( + _JournalStore(objects), broker="ibkr", account_id="DU00000001", + owner_id=profile, profile=profile, intent_id="1"*64, release_id="2"*64, + ) + else: + def make_session(profile): + # Different store instances and equivalent paths share the journal. + path = tmp_path if profile == "first" else tmp_path / "unused" / ".." + return session_at(path, profile=profile) + (tmp_path / "unused").mkdir() + + first, second = make_session("first"), make_session("second") + grant = first.reserve(order(), new_risk=True) + final_read, attempting_open, opened = threading.Event(), threading.Event(), threading.Event() + failures, calls = [], [] + underlying = second._account_mutex + + class NotifyingLock: + def __enter__(self): + attempting_open.set() + return underlying.__enter__() + + def __exit__(self, *args): + return underlying.__exit__(*args) + + object.__setattr__(second, "_account_mutex", NotifyingLock()) + original = AccountSubmissionSession.read_state + reads = [] + + def read(self): + state = original(self) + if self is first: + reads.append(True) + if len(reads) == 2: + final_read.set() + assert attempting_open.wait(5) + return state + + def trip(): + try: + assert final_read.wait(5) + second.record_circuit_breaker_open(snapshot=breached_snapshot()) + opened.set() + except BaseException as exc: + failures.append(exc) + + def submit(): + assert attempting_open.is_set() + assert not opened.is_set() + assert original(second)["circuit_breaker_open"] is False + calls.append(True) + return SimpleNamespace(status="Unknown", broker_order_id=None, filled_quantity=0) + + monkeypatch.setattr(AccountSubmissionSession, "read_state", read) + worker = threading.Thread(target=trip) + worker.start() + try: + first.dispatch(grant, order(), submit) + finally: + worker.join(5) + assert not worker.is_alive() + assert failures == [] + assert calls == [True] + assert opened.is_set() + assert original(second)["circuit_breaker_open"] is True + assert original(first)["attempts"][grant.attempt_id]["certainty"] == "UNKNOWN" + + +def test_failed_final_readback_never_consumes_or_submits_grant(tmp_path, monkeypatch): + first = session_at(tmp_path) + grant = first.reserve(order(), new_risk=True) + original = AccountSubmissionSession.read_state + reads, calls = [], [] + + def read(self): + reads.append(True) + if len(reads) == 2: + raise OSError("synthetic final readback unavailable") + return original(self) + + monkeypatch.setattr(AccountSubmissionSession, "read_state", read) + with pytest.raises(OSError, match="final readback"): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert grant.attempt_id not in first._used + assert original(first)["attempts"][grant.attempt_id]["certainty"] == "PREPARED" From 7f019391b08d09415e78c55cc0d8906a0708df39 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 23:57:58 +0800 Subject: [PATCH 3/4] fix: guard final dispatch across processes Protect local OPEN writes and final dispatch with a shared kernel lock. Require exclusive process ownership for ObjectStore OPEN and dispatch. Preserve UNKNOWN outcomes and the unchanged two-process acceptance test. --- docs/account-new-risk-gate.zh-CN.md | 13 +- .../common/execution_state.py | 114 +++++--- tests/test_cross_process_dispatch.py | 248 ++++++++++++++++++ 3 files changed, 341 insertions(+), 34 deletions(-) create mode 100644 tests/test_cross_process_dispatch.py diff --git a/docs/account-new-risk-gate.zh-CN.md b/docs/account-new-risk-gate.zh-CN.md index 160115f2..e9b4f21f 100644 --- a/docs/account-new-risk-gate.zh-CN.md +++ b/docs/account-new-risk-gate.zh-CN.md @@ -137,10 +137,15 @@ OPEN 则持续为 true，包括其他 profile 及进程重启。恢复观测、C reserve，dispatch 也重新检查 OPEN，阻止已准备但尚未发送的新增风险。省略该参数的 旧调用保持原语义，仍须由各自平台完成类别判断和门控；本次接线范围为 IBKR 的买单。 OPEN 不创建目标、不产生卖单、不替代既有减仓证明，不改变 UNKNOWN 的含义。 -dispatch 完成订单及账户验证后，在同一账户日志的进程内共享锁中重新读回 OPEN， -并持锁调用最终 submit；其他 session 的 OPEN 写入使用同一把锁。锁按日志位置区分， -不随 owner/profile 改变。本锁不提供跨进程或跨主机的分布式互斥；日志排他创建及 -既有账户 owner fence 仍须满足，不能据此声称多个实际写入器已安全协调。 +dispatch 完成订单及账户验证后，使用账户日志的进程内共享锁和跨进程保护， +在保护范围内重新读取 OPEN，并持有保护直到 submit 和结果写入完成；所有 OPEN +写入使用相同保护。锁按日志位置区分，不随 owner/profile 改变。 +本地日志使用同一文件上的内核排他锁，适用于共享同一文件锁域的进程；不能据此 +声称不同主机或不同挂载已协调。ObjectStore 没有锁释放或 CAS 接口，因此使用 +create-only 记录绑定唯一进程，读回确认所有权后才允许 OPEN 写入或 dispatch。 +同一 owner 的其他进程、fork 子进程及重启不能接管该记录；不提供超时、自动清除 +或接管。所有权不明、创建或读回失败时拒绝派发；本地锁失败也拒绝派发。 +已进入 submit 的订单先于等待中的 OPEN，不声称取消在途订单。 这是追加事件类型：旧版本读取新事件会拒绝，不能把混用旧写入器视为安全迁移。 消费者升级须先审核库和平台补丁，再单独协调固定版本、旧写入器停止与运行验证； diff --git a/src/quant_platform_kit/common/execution_state.py b/src/quant_platform_kit/common/execution_state.py index ab1dd841..4616c62a 100644 --- a/src/quant_platform_kit/common/execution_state.py +++ b/src/quant_platform_kit/common/execution_state.py @@ -11,6 +11,7 @@ import re import tempfile from collections.abc import Callable, Mapping +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from fractions import Fraction @@ -761,6 +762,7 @@ def _date_key(value: object) -> str: _SUBMISSION_TERMINAL = frozenset({"Filled", "Rejected", "rejected", "Cancelled", "ApiCancelled"}) _ACCOUNT_SUBMISSION_LOCKS: dict[str, Any] = {} _ACCOUNT_SUBMISSION_LOCKS_MUTEX = threading.Lock() +_ACCOUNT_WRITER_TOKEN = uuid.uuid4().hex class AccountSubmissionError(RuntimeError): @@ -868,6 +870,49 @@ def __init__(self, store: ExecutionMarkerStore, *, broker: str, account_id: str, account_mutex = _ACCOUNT_SUBMISSION_LOCKS.setdefault(lock_key, threading.RLock()) object.__setattr__(self, "_account_mutex", account_mutex) + @contextmanager + def _writer_guard(self): + """Exclude OPEN writers through submit, or deny when exclusion is unknown. + + Local journals use the same kernel lock in every cooperating process. + ObjectStore has no release/CAS contract: claim permanent process ownership + instead. No lease expiry or automatic takeover can authorize a restart. + """ + if self.store.cloud_prefix_uri: + location = self.store._cloud_uri_for_namespace( + self.account_root, "account_writer_ownership", + ) + # PID distinguishes forked children, which inherit the random token. + token = f"{_ACCOUNT_WRITER_TOKEN}:{os.getpid()}" + payload = json.dumps({"kind": "WRITER_OWNERSHIP", "token": token}) + try: + objects = self.store._object_store() + created = objects.create_text(location, payload, "application/json") + if created is not True and created is not False: + raise ValueError("unknown create result") + if objects.read_text(location) != payload: + raise ValueError("different writer") + except Exception as exc: + raise AccountSubmissionError("exclusive account writer not established") from exc + yield + return + fd = None + try: + import fcntl + location = self._location().resolve() / "writer.lock" + location.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(location, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + fcntl.flock(fd, fcntl.LOCK_EX) + except (ImportError, OSError): + if fd is not None: + os.close(fd) + object.__setattr__(self, "_failed", True) + raise + try: + yield + finally: + os.close(fd) + def _location(self, sequence: int | None = None): suffix = self.account_root + "/" + (f"{sequence:020d}.json" if sequence is not None else "") if self.store.cloud_prefix_uri: @@ -1177,7 +1222,7 @@ def record_circuit_breaker_open(self, *, snapshot): """ from quant_platform_kit.risk.account_new_risk_gate import evaluate_new_risk_admission from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope - with self._mutex, self._account_mutex: + with self._mutex, self._account_mutex, self._writer_guard(): if self._failed: raise AccountSubmissionError("session persistence uncertain") if not evaluate_new_risk_admission(snapshot).hard_brake_triggered: @@ -1601,38 +1646,47 @@ def dispatch(self, grant: AccountSubmissionGrant, order, submit: Callable[[], An raise AccountSubmissionError("grant no longer prepared") if attempt.get("new_risk") and current["circuit_breaker_open"]: raise AccountSubmissionError("new risk prohibited by circuit breaker") - self._used.add(grant.attempt_id) - try: - report = submit() - except Exception as exc: - try: - self.record_observation(grant, None) - except Exception as persist_exc: - exc.add_note("account outcome persistence also failed: " + type(persist_exc).__name__) - object.__setattr__(self, "_failed", True) - raise - # Keep the outcome append in the same critical section so an - # OPEN writer cannot race its next sequence with OBSERVED. - if record_observation_watermark: + # Acquire cross-process exclusion after the preliminary read; + # only a read inside this guard may authorize the broker write. + with self._writer_guard(): + current = self._owned_grant(grant) + attempt = current["attempts"][grant.attempt_id] + if attempt["certainty"] != "PREPARED": + raise AccountSubmissionError("grant no longer prepared") + if attempt.get("new_risk") and current["circuit_breaker_open"]: + raise AccountSubmissionError("new risk prohibited by circuit breaker") + self._used.add(grant.attempt_id) try: - now = self.clock() - recorded_at = now.isoformat() if isinstance(now, datetime) else now - from quant_platform_kit.common.account_domain import _time - _time(recorded_at) - except Exception: - recorded_at = None - if recorded_at is not None: - self.record_observation(grant, report, recorded_at=recorded_at) + report = submit() + except Exception as exc: + try: + self.record_observation(grant, None) + except Exception as persist_exc: + exc.add_note("account outcome persistence also failed: " + type(persist_exc).__name__) + object.__setattr__(self, "_failed", True) + raise + # Keep the outcome append in the same critical section so an + # OPEN writer cannot race its next sequence with OBSERVED. + if record_observation_watermark: + try: + now = self.clock() + recorded_at = now.isoformat() if isinstance(now, datetime) else now + from quant_platform_kit.common.account_domain import _time + _time(recorded_at) + except Exception: + recorded_at = None + if recorded_at is not None: + self.record_observation(grant, report, recorded_at=recorded_at) + else: + self.record_observation(grant, report) else: self.record_observation(grant, report) - else: - self.record_observation(grant, report) - attempt = self.read_state()["attempts"][grant.attempt_id] - if attempt["certainty"] == "UNKNOWN" and str(getattr(report, "status", "")).strip() in ( - _SUBMISSION_ACK | _SUBMISSION_PARTIAL | _SUBMISSION_TERMINAL - ): - raise AccountSubmissionError("broker result facts are uncertain; original attempt retained") - return report + attempt = self.read_state()["attempts"][grant.attempt_id] + if attempt["certainty"] == "UNKNOWN" and str(getattr(report, "status", "")).strip() in ( + _SUBMISSION_ACK | _SUBMISSION_PARTIAL | _SUBMISSION_TERMINAL + ): + raise AccountSubmissionError("broker result facts are uncertain; original attempt retained") + return report def record_observation(self, grant: AccountSubmissionGrant, report, *, recorded_at=None) -> bool: """Persist original facts. Exact replay is idempotent, never a new grant.""" diff --git a/tests/test_cross_process_dispatch.py b/tests/test_cross_process_dispatch.py new file mode 100644 index 00000000..3a646528 --- /dev/null +++ b/tests/test_cross_process_dispatch.py @@ -0,0 +1,248 @@ +"""Independent synthetic acceptance checks. No broker or cloud access.""" +from pathlib import Path +import subprocess +import sys +from types import SimpleNamespace + +from quant_platform_kit.common.execution_state import ( + AccountSubmissionError, AccountSubmissionSession, ExecutionMarkerStore, + claim_account_owner, +) +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot + + +def session(path): + return AccountSubmissionSession( + ExecutionMarkerStore(local_dir=path), broker="ibkr", account_id="DU00000001", + owner_id="synthetic-owner", profile="synthetic-profile", + intent_id="1" * 64, release_id="2" * 64, + ) + + +def order(): + return SimpleNamespace(account_id="DU00000001", symbol="TQQQ", side="buy", quantity=1) + + +def test_cross_process_open_after_final_read_must_prevent_submit(tmp_path, monkeypatch): + """A scheduling pause after final read must not permit post-OPEN new risk.""" + first = session(tmp_path) + assert claim_account_owner(first.store, broker=first.broker, + account_id=first.account_id, owner_id=first.owner_id).allowed + grant = first.reserve(order(), new_risk=True) + original = AccountSubmissionSession._owned_grant + reads, calls, opened = [], [], [] + source = Path(sys.modules[AccountSubmissionSession.__module__].__file__).resolve().parents[2] + child = ''' +import sys +sys.path.insert(0, sys.argv[1]) +from quant_platform_kit.common.execution_state import AccountSubmissionSession, ExecutionMarkerStore, claim_account_owner +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot +s = AccountSubmissionSession(ExecutionMarkerStore(local_dir=sys.argv[2]), broker="ibkr", account_id="DU00000001", owner_id="synthetic-owner", profile="synthetic-profile", intent_id="3"*64, release_id="4"*64) +assert claim_account_owner(s.store, broker=s.broker, account_id=s.account_id, owner_id=s.owner_id).allowed +s.record_circuit_breaker_open(snapshot=InjectedReconciliationSnapshot(observation_status="COMPLETE", reconciliation_status="VERIFIED", circuit_breaker_state="CLOSED", equity_usd=40000, peak_equity_usd=50000, drawdown_from_peak=.2)) +assert s.read_state()["circuit_breaker_open"] is True +print("OPEN durably confirmed; same-owner fence allowed") +''' + + def pause_after_final_read(self, candidate): + state = original(self, candidate) + if self is first: + reads.append(True) + if len(reads) == 2: + # Only inject a scheduling pause; use the real read, append, + # readback, owner fence and a fresh interpreter in the child. + result = subprocess.run([sys.executable, "-c", child, str(source), str(tmp_path)], + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, result.stderr + opened.append(result.stdout.strip()) + return state + + def submit(): + actual = session(tmp_path).read_state() + calls.append(actual["circuit_breaker_open"]) + return SimpleNamespace(status="Unknown", broker_order_id=None, filled_quantity=0) + + monkeypatch.setattr(AccountSubmissionSession, "_owned_grant", pause_after_final_read) + error = None + try: + first.dispatch(grant, order(), submit) + except AccountSubmissionError as exc: + error = str(exc) + assert opened == ["OPEN durably confirmed; same-owner fence allowed"] + state = first.read_state() + assert state["circuit_breaker_open"] is True + assert calls == [], ( + f"submit executed with current OPEN={calls}; dispatch error={error!r}; " + f"final certainty={state['attempts'][grant.attempt_id]['certainty']}" + ) + + +def test_local_lock_covers_submit_and_open_in_another_process(tmp_path): + import time + + first = session(tmp_path) + grant = first.reserve(order(), new_risk=True) + repo = Path(__file__).resolve().parents[1] + child = ''' +import sys, fcntl +from pathlib import Path +sys.path.insert(0, sys.argv[1]) +sys.path.insert(0, sys.argv[1] + "/src") +from tests.test_cross_process_dispatch import session +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot +s = session(Path(sys.argv[2])) +original = fcntl.flock +def lock(fd, operation): + Path(sys.argv[3]).write_text("attempting") + return original(fd, operation) +fcntl.flock = lock +s.record_circuit_breaker_open(snapshot=InjectedReconciliationSnapshot(observation_status="COMPLETE", reconciliation_status="VERIFIED", circuit_breaker_state="CLOSED", equity_usd=40000, peak_equity_usd=50000, drawdown_from_peak=.2)) +assert s.read_state()["circuit_breaker_open"] is True +print("OPEN confirmed") +''' + children = [] + + def submit(): + ready = tmp_path / 'ready' + process = subprocess.Popen( + [sys.executable, '-c', child, str(repo), str(tmp_path), str(ready)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + children.append(process) + deadline = time.monotonic() + 10 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(.01) + assert ready.exists() + assert process.poll() is None + assert first.read_state()['circuit_breaker_open'] is False + return SimpleNamespace(status='Unknown', broker_order_id=None, filled_quantity=0) + + first.dispatch(grant, order(), submit) + stdout, stderr = children[0].communicate(timeout=20) + assert children[0].returncode == 0, stderr + assert stdout.strip() == 'OPEN confirmed' + state = first.read_state() + assert state['circuit_breaker_open'] is True + assert state['attempts'][grant.attempt_id]['certainty'] == 'UNKNOWN' + + +class _FileObjects: + """Synthetic shared ObjectStore for independent-process ownership checks.""" + prefix = 'gs://synthetic-bucket/test/' + + def __init__(self, root): + self.root = Path(root) + + def path(self, uri): + assert uri.startswith(self.prefix) + return self.root / uri.removeprefix(self.prefix) + + def list(self, prefix): + return [self.prefix + str(p.relative_to(self.root)) + for p in self.root.rglob('*.json') + if (self.prefix + str(p.relative_to(self.root))).startswith(prefix)] + + def read_text(self, uri): + return self.path(uri).read_text() + + def create_text(self, uri, text, content_type): + path = self.path(uri) + path.parent.mkdir(parents=True, exist_ok=True) + try: + with path.open('x') as handle: + handle.write(text) + except FileExistsError: + return False + return True + + +class _FileStore(ExecutionMarkerStore): + def _object_store(self): + return _FileObjects(self.local_dir) + + +def cloud_session(path, intent='1'): + return AccountSubmissionSession( + _FileStore(local_dir=path, cloud_prefix_uri='gs://synthetic-bucket/test'), + broker='ibkr', account_id='DU00000001', owner_id='synthetic-owner', + profile='synthetic-profile', intent_id=intent * 64, release_id='2' * 64, + ) + + +def test_cloud_exclusive_process_owner_denies_same_owner_restart(tmp_path): + first = cloud_session(tmp_path) + with first._account_mutex, first._writer_guard(): + pass + child = ''' +import sys +sys.path.insert(0, sys.argv[1]) +sys.path.insert(0, sys.argv[1] + "/src") +from tests.test_cross_process_dispatch import cloud_session, order +from quant_platform_kit.common.execution_state import AccountSubmissionError +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot +s = cloud_session(sys.argv[2], intent="3") +g = s.reserve(order(), new_risk=True) +calls = [] +try: + s.dispatch(g, order(), lambda: calls.append(True)) +except AccountSubmissionError as exc: + assert "exclusive account writer" in str(exc) +else: + raise AssertionError("restart dispatched") +assert calls == [] +assert g.attempt_id not in s._used +assert s.read_state()["attempts"][g.attempt_id]["certainty"] == "PREPARED" +try: + s.record_circuit_breaker_open(snapshot=InjectedReconciliationSnapshot(observation_status="COMPLETE", reconciliation_status="VERIFIED", circuit_breaker_state="CLOSED", equity_usd=40000, peak_equity_usd=50000, drawdown_from_peak=.2)) +except AccountSubmissionError: + pass +else: + raise AssertionError("other process wrote OPEN") +print("restart denied") +''' + result = subprocess.run([sys.executable, '-c', child, + str(Path(__file__).resolve().parents[1]), str(tmp_path)], + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == 'restart denied' + assert first.read_state()['circuit_breaker_open'] is False + + +def test_cloud_ownership_readback_failure_never_consumes_or_submits(tmp_path, monkeypatch): + import pytest + + first = cloud_session(tmp_path) + grant = first.reserve(order(), new_risk=True) + original = _FileObjects.read_text + + def read(self, uri): + if 'account_writer_ownership/' in uri: + raise OSError('synthetic ownership read failure') + return original(self, uri) + + monkeypatch.setattr(_FileObjects, 'read_text', read) + calls = [] + with pytest.raises(AccountSubmissionError, match='exclusive account writer'): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert grant.attempt_id not in first._used + assert first.read_state()['attempts'][grant.attempt_id]['certainty'] == 'PREPARED' + + +def test_local_lock_failure_never_consumes_or_submits(tmp_path, monkeypatch): + import fcntl + import pytest + + first = session(tmp_path) + grant = first.reserve(order(), new_risk=True) + + def fail(*args): + raise OSError('synthetic locking unavailable') + + monkeypatch.setattr(fcntl, 'flock', fail) + calls = [] + with pytest.raises(OSError, match='locking unavailable'): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert first._failed + assert grant.attempt_id not in first._used From f8cf8109867cbe4bb172d0cc1d300984ae77854f Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 3 Nov 2026 00:33:11 +0800 Subject: [PATCH 4/4] fix(risk): Preserve breaker evidence and verify writer identity --- docs/account-new-risk-gate.zh-CN.md | 9 + .../common/execution_state.py | 80 ++++- tests/test_breaker_lock_identity.py | 289 ++++++++++++++++++ 3 files changed, 365 insertions(+), 13 deletions(-) create mode 100644 tests/test_breaker_lock_identity.py diff --git a/docs/account-new-risk-gate.zh-CN.md b/docs/account-new-risk-gate.zh-CN.md index e9b4f21f..4fdf8706 100644 --- a/docs/account-new-risk-gate.zh-CN.md +++ b/docs/account-new-risk-gate.zh-CN.md @@ -127,6 +127,8 @@ W1 平台 adapter 已有实现；持久化 OPEN 熔断、实际账户读回和 事件只含注入的 equity/peak/drawdown 与调用方身份，不证明其真实来源或资金流口径。 显式 drawdown 无效但有效 equity/peak 已独立确认越界时，日志保留 equity/peak， 将 drawdown 记为 null，由读回重新计算回撤并验证 OPEN；无效值不能使该次越界丢失。 +有效的显式 drawdown 已确认越界而补充 peak 无效时，保留 equity/drawdown， +将 peak 记为 null；读回仍须用保留的有效事实验证 OPEN。 原始观测的数据错误仍由风险门报告，不把无效值当作有效事实。 `read_state()` 增加 `circuit_breaker_open` 布尔字段。合法旧日志初始为 false；已记录 @@ -145,6 +147,13 @@ dispatch 完成订单及账户验证后，使用账户日志的进程内共享 create-only 记录绑定唯一进程，读回确认所有权后才允许 OPEN 写入或 dispatch。 同一 owner 的其他进程、fork 子进程及重启不能接管该记录；不提供超时、自动清除 或接管。所有权不明、创建或读回失败时拒绝派发；本地锁失败也拒绝派发。 +保护建立后，最终日志读取后及 submit 紧前再次确认身份：本地核对文件描述符与 +锁路径的设备号/inode，拒绝路径丢失、替换或符号链接；ObjectStore 重新读回并 +核对进程所有权。已确认的 session 身份不能通过重建丢失的锁或记录来恢复。 +检测失败使 session 停止写入；已消耗的尝试继续保持未决，不能按未成交重试。 +submit 返回后与保护结束前也检查身份；发生在已进入 submit 之后的丢失不能撤销 +券商动作。存储须保证锁路径稳定、所有权记录不可变及读回一致；检查不代替这些 +约束，也不能证明两次检查之间任意外部存储修改或旧写入器已经受控。 已进入 submit 的订单先于等待中的 OPEN，不声称取消在途订单。 这是追加事件类型：旧版本读取新事件会拒绝，不能把混用旧写入器视为安全迁移。 diff --git a/src/quant_platform_kit/common/execution_state.py b/src/quant_platform_kit/common/execution_state.py index 4616c62a..42be57d9 100644 --- a/src/quant_platform_kit/common/execution_state.py +++ b/src/quant_platform_kit/common/execution_state.py @@ -9,6 +9,7 @@ import hashlib import os import re +import stat import tempfile from collections.abc import Callable, Mapping from contextlib import contextmanager @@ -856,6 +857,7 @@ def __init__(self, store: ExecutionMarkerStore, *, broker: str, account_id: str, "intent_id": _submission_digest(intent_id), "release_id": _submission_digest(release_id), "account_root": "v1/" + submission_binding_digest({"broker": broker, "account_id": account}), "_issued": {}, "_used": set(), "_mutex": threading.RLock(), "_failed": False, + "_writer_identity": None, "account_domain": account_domain, "observed_account_id": observed_account_id, "broker_identity": broker_identity, "observed_at": observed_at, "decision_at": decision_at, "observed_environment": observed_environment, "observed_server": observed_server, @@ -878,6 +880,8 @@ def _writer_guard(self): ObjectStore has no release/CAS contract: claim permanent process ownership instead. No lease expiry or automatic takeover can authorize a restart. """ + if self._failed: + raise AccountSubmissionError("session persistence uncertain") if self.store.cloud_prefix_uri: location = self.store._cloud_uri_for_namespace( self.account_root, "account_writer_ownership", @@ -885,31 +889,70 @@ def _writer_guard(self): # PID distinguishes forked children, which inherit the random token. token = f"{_ACCOUNT_WRITER_TOKEN}:{os.getpid()}" payload = json.dumps({"kind": "WRITER_OWNERSHIP", "token": token}) + identity = (location, payload) + + def confirm_writer(): + try: + if objects.read_text(location) != payload: + raise ValueError("different writer") + except Exception as exc: + object.__setattr__(self, "_failed", True) + raise AccountSubmissionError("exclusive account writer identity lost or unknown") from exc + try: objects = self.store._object_store() - created = objects.create_text(location, payload, "application/json") - if created is not True and created is not False: - raise ValueError("unknown create result") - if objects.read_text(location) != payload: - raise ValueError("different writer") + if self._writer_identity is None: + created = objects.create_text(location, payload, "application/json") + if created is not True and created is not False: + raise ValueError("unknown create result") + elif self._writer_identity != identity: + raise ValueError("writer identity changed") + confirm_writer() except Exception as exc: + object.__setattr__(self, "_failed", True) raise AccountSubmissionError("exclusive account writer not established") from exc - yield + object.__setattr__(self, "_writer_identity", identity) + yield confirm_writer + confirm_writer() return fd = None try: import fcntl location = self._location().resolve() / "writer.lock" location.parent.mkdir(parents=True, exist_ok=True) - fd = os.open(location, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + # Creation is initial setup only. A known lock must never be + # recreated: another process may still hold its detached inode. + flags = os.O_RDWR | os.O_NOFOLLOW + if self._writer_identity is None: + flags |= os.O_CREAT + fd = os.open(location, flags, 0o600) + acquired = os.fstat(fd) + identity = (str(location), acquired.st_dev, acquired.st_ino) + if self._writer_identity is not None and self._writer_identity != identity: + raise AccountSubmissionError("exclusive account writer identity changed") fcntl.flock(fd, fcntl.LOCK_EX) - except (ImportError, OSError): + except (ImportError, OSError, AccountSubmissionError): if fd is not None: os.close(fd) object.__setattr__(self, "_failed", True) raise try: - yield + def confirm_writer(): + try: + held = os.fstat(fd) + named = os.stat(location, follow_symlinks=False) + if (not stat.S_ISREG(named.st_mode) or held.st_nlink != 1 + or (held.st_dev, held.st_ino) != identity[1:] + or (named.st_dev, named.st_ino) != identity[1:]): + raise ValueError("lock pathname no longer names the locked inode") + except (OSError, ValueError) as exc: + object.__setattr__(self, "_failed", True) + raise AccountSubmissionError("exclusive account writer identity lost or unknown") from exc + + confirm_writer() + object.__setattr__(self, "_writer_identity", identity) + yield confirm_writer + confirm_writer() finally: os.close(fd) @@ -1222,13 +1265,19 @@ def record_circuit_breaker_open(self, *, snapshot): """ from quant_platform_kit.risk.account_new_risk_gate import evaluate_new_risk_admission from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope - with self._mutex, self._account_mutex, self._writer_guard(): + with self._mutex, self._account_mutex, self._writer_guard() as confirm_writer: if self._failed: raise AccountSubmissionError("session persistence uncertain") if not evaluate_new_risk_admission(snapshot).hard_brake_triggered: raise AccountSubmissionError("circuit breaker OPEN requires a drawdown breach") - data = {key: None if getattr(snapshot, key) is None else _submission_number(getattr(snapshot, key)) - for key in ("equity_usd", "peak_equity_usd")} + peak = snapshot.peak_equity_usd + # A supplemental input error cannot erase an independently valid + # explicit drawdown breach. Replay validates the remaining facts. + if peak is not None and (isinstance(peak, bool) or not isinstance(peak, (int, float)) + or not math.isfinite(float(peak)) or peak <= 0): + peak = None + data = {"equity_usd": _submission_number(snapshot.equity_usd), + "peak_equity_usd": None if peak is None else _submission_number(peak)} drawdown = snapshot.drawdown_from_peak if "INVALID_DRAWDOWN_FAIL_CLOSED" in evaluate_capital_risk_envelope( snapshot.equity_usd, drawdown_from_peak=drawdown, @@ -1239,6 +1288,7 @@ def record_circuit_breaker_open(self, *, snapshot): data["drawdown_from_peak"] = None if drawdown is None else _submission_number(drawdown) data.update(owner_id=self.owner_id, profile=self.profile) state = self.read_state() + confirm_writer() if not state["circuit_breaker_open"]: self._append(state, "CIRCUIT_BREAKER_OPEN", data) @@ -1648,23 +1698,27 @@ def dispatch(self, grant: AccountSubmissionGrant, order, submit: Callable[[], An raise AccountSubmissionError("new risk prohibited by circuit breaker") # Acquire cross-process exclusion after the preliminary read; # only a read inside this guard may authorize the broker write. - with self._writer_guard(): + with self._writer_guard() as confirm_writer: current = self._owned_grant(grant) attempt = current["attempts"][grant.attempt_id] if attempt["certainty"] != "PREPARED": raise AccountSubmissionError("grant no longer prepared") if attempt.get("new_risk") and current["circuit_breaker_open"]: raise AccountSubmissionError("new risk prohibited by circuit breaker") + confirm_writer() self._used.add(grant.attempt_id) try: + confirm_writer() report = submit() except Exception as exc: try: + confirm_writer() self.record_observation(grant, None) except Exception as persist_exc: exc.add_note("account outcome persistence also failed: " + type(persist_exc).__name__) object.__setattr__(self, "_failed", True) raise + confirm_writer() # Keep the outcome append in the same critical section so an # OPEN writer cannot race its next sequence with OBSERVED. if record_observation_watermark: diff --git a/tests/test_breaker_lock_identity.py b/tests/test_breaker_lock_identity.py new file mode 100644 index 00000000..bdc902f8 --- /dev/null +++ b/tests/test_breaker_lock_identity.py @@ -0,0 +1,289 @@ +"""Reviewer-only adverse cases; synthetic local storage and no network.""" +from dataclasses import replace +from pathlib import Path +import subprocess +import sys +from types import SimpleNamespace + +import pytest + +from quant_platform_kit.common.execution_state import ( + AccountSubmissionSession, AccountSubmissionError, ExecutionMarkerStore, +) +from quant_platform_kit.risk.account_new_risk_gate import ( + InjectedReconciliationSnapshot, evaluate_new_risk_admission, +) +from tests.test_cross_process_dispatch import cloud_session, _FileObjects + + +def session(path): + return AccountSubmissionSession(ExecutionMarkerStore(local_dir=path), + broker='ibkr', account_id='DU00000001', owner_id='synthetic-owner', + profile='synthetic-profile', intent_id='1'*64, release_id='2'*64) + + +def order(): + return SimpleNamespace(account_id='DU00000001', symbol='TQQQ', side='buy', quantity=1) + + +def breach(): + return InjectedReconciliationSnapshot(observation_status='COMPLETE', + reconciliation_status='VERIFIED', circuit_breaker_state='CLOSED', + equity_usd=40000., peak_equity_usd=50000., drawdown_from_peak=.2) + + +@pytest.mark.parametrize('peak', [float('nan'), float('inf'), -1., True, 'unknown']) +def test_valid_explicit_breach_with_invalid_peak_is_durable(tmp_path, peak): + snapshot = replace(breach(), peak_equity_usd=peak) + admission = evaluate_new_risk_admission(snapshot) + assert admission.hard_brake_triggered and admission.risk_input_invalid + first = session(tmp_path) + error = None + try: + first.record_circuit_breaker_open(snapshot=snapshot) + except AccountSubmissionError as exc: + error = str(exc) + source = Path(sys.modules[AccountSubmissionSession.__module__].__file__).resolve().parents[2] + child = ''' +import sys, json +from types import SimpleNamespace +sys.path.insert(0, sys.argv[1]) +from quant_platform_kit.common.execution_state import AccountSubmissionSession, AccountSubmissionError, ExecutionMarkerStore +s=AccountSubmissionSession(ExecutionMarkerStore(local_dir=sys.argv[2]),broker='ibkr',account_id='DU00000001',owner_id='synthetic-owner',profile='synthetic-profile',intent_id='3'*64,release_id='4'*64) +state=s.read_state() +admitted=False +try: + s.reserve(SimpleNamespace(account_id='DU00000001',symbol='TQQQ',side='buy',quantity=1),new_risk=True) + admitted=True +except AccountSubmissionError: pass +print(json.dumps({'open':state['circuit_breaker_open'],'generation':state['generation'],'admitted':admitted})) +''' + result = subprocess.run([sys.executable, '-c', child, str(source), str(tmp_path)], + capture_output=True, text=True, timeout=10) + assert result.returncode == 0, result.stderr + import json + state = json.loads(result.stdout) + assert state['open'] and not state['admitted'], (error, state) + + +def test_replaced_local_lock_blocks_dispatch(tmp_path, monkeypatch): + first = session(tmp_path) + grant = first.reserve(order(), new_risk=True) + original = AccountSubmissionSession._owned_grant + reads, calls = [], [] + source = Path(sys.modules[AccountSubmissionSession.__module__].__file__).resolve().parents[2] + child = ''' +import sys +sys.path.insert(0, sys.argv[1]) +from quant_platform_kit.common.execution_state import AccountSubmissionSession, ExecutionMarkerStore +from quant_platform_kit.risk.account_new_risk_gate import InjectedReconciliationSnapshot +s=AccountSubmissionSession(ExecutionMarkerStore(local_dir=sys.argv[2]),broker='ibkr',account_id='DU00000001',owner_id='synthetic-owner',profile='synthetic-profile',intent_id='3'*64,release_id='4'*64) +s.record_circuit_breaker_open(snapshot=InjectedReconciliationSnapshot(observation_status='COMPLETE',reconciliation_status='VERIFIED',circuit_breaker_state='CLOSED',equity_usd=40000,peak_equity_usd=50000,drawdown_from_peak=.2)) +assert s.read_state()['circuit_breaker_open'] +print('OPEN confirmed') +''' + def after_read(self, candidate): + state = original(self, candidate) + if self is first: + reads.append(True) + if len(reads) == 3: # Real final read under acquired flock. + path = first._location().resolve() / 'writer.lock' + # Simulate loss of the pathname's lock identity, retaining all + # journal data and the original locked inode for inspection. + path.rename(path.with_name('detached.lock')) + result = subprocess.run([sys.executable, '-c', child, str(source), str(tmp_path)], + capture_output=True, text=True, timeout=10) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == 'OPEN confirmed' + return state + monkeypatch.setattr(AccountSubmissionSession, '_owned_grant', after_read) + def submit(): + calls.append(first.read_state()['circuit_breaker_open']) + return SimpleNamespace(status='Unknown', broker_order_id=None, filled_quantity=0) + try: + first.dispatch(grant, order(), submit) + except (AccountSubmissionError, OSError): + pass + assert calls == [], f'submit ran after loss of original lock identity, OPEN={calls}' + + +def test_lost_cloud_ownership_before_final_read_blocks_dispatch(tmp_path, monkeypatch): + first = cloud_session(tmp_path) + grant = first.reserve(order(), new_risk=True) + original = AccountSubmissionSession._owned_grant + reads, calls = [], [] + def before_read(self, candidate): + if self is first: + reads.append(True) + if len(reads) == 3: + uri = first.store._cloud_uri_for_namespace(first.account_root, 'account_writer_ownership') + # Storage fault/revocation after initial ownership confirmation. + _FileObjects(tmp_path).path(uri).write_text('{"kind":"WRITER_OWNERSHIP","token":"synthetic-other-process"}') + return original(self, candidate) + monkeypatch.setattr(AccountSubmissionSession, '_owned_grant', before_read) + try: + first.dispatch(grant, order(), lambda: calls.append(True)) + except (AccountSubmissionError, OSError): + pass + assert calls == [], 'submit ran after exclusive ownership was lost' + + +@pytest.mark.parametrize('factory', [session, cloud_session]) +@pytest.mark.parametrize('peak', [0., False, '50000', '', float('-inf')]) +def test_invalid_peak_is_omitted_without_clearing_input_error(tmp_path, factory, peak): + import json + + snapshot = replace(breach(), peak_equity_usd=peak) + first = factory(tmp_path) + first.record_circuit_breaker_open(snapshot=snapshot) + admission = evaluate_new_risk_admission(snapshot) + assert admission.hard_brake_triggered and admission.risk_input_invalid + if first.store.cloud_prefix_uri: + event = json.loads(_FileObjects(tmp_path).read_text(first._location(1))) + else: + event = json.loads(first._location(1).read_text()) + assert event['data']['peak_equity_usd'] is None + assert event['data']['drawdown_from_peak'] == .2 + second = factory(tmp_path) + assert second.read_state()['circuit_breaker_open'] is True + with pytest.raises(AccountSubmissionError): + second.reserve(order(), new_risk=True) + + +@pytest.mark.parametrize('factory,fault', [ + (factory, fault) for factory in (session, cloud_session) + for fault in ('missing', 'replacement', 'symlink', 'unreadable') + if factory is session or fault != 'symlink' +]) +def test_identity_loss_after_final_read_fails_closed(tmp_path, monkeypatch, factory, fault): + first = factory(tmp_path) + grant = first.reserve(order(), new_risk=True) + original = AccountSubmissionSession._owned_grant + reads, calls = [], [] + + def after_read(self, candidate): + state = original(self, candidate) + if self is first: + reads.append(True) + if len(reads) == 3: + if factory is cloud_session: + uri = first.store._cloud_uri_for_namespace(first.account_root, 'account_writer_ownership') + path = _FileObjects(tmp_path).path(uri) + else: + path = first._location().resolve() / 'writer.lock' + if fault == 'missing': + path.unlink() + elif fault in {'replacement', 'symlink'}: + detached = path.with_name('detached.lock') + path.rename(detached) + if fault == 'symlink': + path.symlink_to(detached) + else: + path.write_text('synthetic replacement') + elif factory is cloud_session: + original_read = _FileObjects.read_text + def read(objects, location): + if location == uri: + raise OSError('synthetic ownership unavailable') + return original_read(objects, location) + monkeypatch.setattr(_FileObjects, 'read_text', read) + else: + import os + original_stat = os.stat + def stat(location, *args, **kwargs): + if Path(location) == path: + raise OSError('synthetic lock identity unavailable') + return original_stat(location, *args, **kwargs) + monkeypatch.setattr(os, 'stat', stat) + return state + + monkeypatch.setattr(AccountSubmissionSession, '_owned_grant', after_read) + with pytest.raises((AccountSubmissionError, OSError)): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert first._failed + assert grant.attempt_id not in first._used + assert first.read_state()['attempts'][grant.attempt_id]['certainty'] == 'PREPARED' + with pytest.raises(AccountSubmissionError): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + + +@pytest.mark.parametrize('factory', [session, cloud_session]) +def test_missing_established_identity_is_not_recreated(tmp_path, factory): + first = factory(tmp_path) + with first._writer_guard(): + pass + grant = first.reserve(order(), new_risk=True) + if factory is cloud_session: + uri = first.store._cloud_uri_for_namespace(first.account_root, 'account_writer_ownership') + path = _FileObjects(tmp_path).path(uri) + else: + path = first._location().resolve() / 'writer.lock' + path.unlink() + calls = [] + with pytest.raises((AccountSubmissionError, OSError)): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert not path.exists() + assert first._failed + + +@pytest.mark.parametrize('factory', [session, cloud_session]) +@pytest.mark.parametrize('boundary', ['consume', 'submit']) +def test_identity_loss_at_submit_boundary_retains_unresolved_attempt(tmp_path, factory, boundary): + first = factory(tmp_path) + grant = first.reserve(order(), new_risk=True) + if factory is cloud_session: + uri = first.store._cloud_uri_for_namespace(first.account_root, 'account_writer_ownership') + path = _FileObjects(tmp_path).path(uri) + else: + path = first._location().resolve() / 'writer.lock' + calls = [] + + class FaultingSet(set): + def add(self, value): + super().add(value) + path.unlink() + + if boundary == 'consume': + object.__setattr__(first, '_used', FaultingSet()) + + def submit(): + calls.append(True) + path.unlink() + return SimpleNamespace(status='Filled', broker_order_id='synthetic-order', + filled_quantity=1, average_fill_price=100) + + with pytest.raises((AccountSubmissionError, OSError)): + first.dispatch(grant, order(), submit) + assert calls == ([] if boundary == 'consume' else [True]) + assert first._failed + assert grant.attempt_id in first._used + state = first.read_state() + assert state['attempts'][grant.attempt_id]['certainty'] == 'PREPARED' + assert state['attempts'][grant.attempt_id]['observation'] is None + with pytest.raises(AccountSubmissionError): + first.dispatch(grant, order(), submit) + + +def test_lock_replaced_during_acquisition_cannot_authorize_submit(tmp_path, monkeypatch): + import fcntl + + first = session(tmp_path) + grant = first.reserve(order(), new_risk=True) + lock = first._location().resolve() / 'writer.lock' + original = fcntl.flock + + def acquire(fd, operation): + original(fd, operation) + lock.rename(lock.with_name('detached.lock')) + lock.write_text('synthetic replacement') + + monkeypatch.setattr(fcntl, 'flock', acquire) + calls = [] + with pytest.raises((AccountSubmissionError, OSError)): + first.dispatch(grant, order(), lambda: calls.append(True)) + assert calls == [] + assert first._failed + assert grant.attempt_id not in first._used