Activity

1. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   {"status":"blocked","reason":"无法访问本地 checkout：sandbox 初始化失败（RTM_NEWADDR: Operation not permitted），因此无法可靠返回两份完整文件内容或执行 pinned offline tests。未修改任何文件、未执行测试、未进行部署或 broker 操作。"}
2. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   Codex Audit

   Codex execution failed with exit code 1.

   No files were pushed. Check the bridge workflow logs for details.
3. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   结论：问题已在提供的快照中得到确认，属于“已成功认领但重复证据分支未收敛命令状态”的状态机缺陷。它不会直接证明重复下单，但会让后续周期持续被旧的 CLAIMED 命令阻塞。

   关键证据：

   • run_strategy() 先对 due command 调用 claim_due()，设置 live_command_claimed=True。
   • 随后才检查 execution_state_store.has_marker() 和 has_prior_execution_report()。
   • 若发现重复证据，进入 execution_already_recorded 分支，构造 action_done=False 的结果。
   • 该分支没有调用 execution_command_store.append_event()，因此当前命令仍保持 CLAIMED。
   • 这与 issue 中 SG 9/11 的历史证据完全一致：无 broker submission、无订单，但后续仍被 unresolved command 阻塞。

   风险判断：High。主要影响是持续阻塞后续 durable live commands；现有保护逻辑不会因此允许重复下单。

   建议的最小修复方向：

   1. 仅在“当前命令已 claim、重复证据与当前命令的 platform/account/strategy/signal/effective 绑定一致、且明确证明没有新的 broker submission”时，将当前 claim 写入现有终态，例如 CANCELLED，并记录明确原因 duplicate_execution_confirmed。
   2. 必须使用 expected_previous_state=CLAIMED，避免并发状态覆盖。
   3. has_marker() 或 prior-report 读取失败、证据无法区分普通 claim 与已完成执行、身份不匹配、存在 pending/unknown order 时，不写终态，继续保持 CLAIMED。
   4. 回归测试应覆盖：
      • 历史 SG 日期与 command ID 的正向重复路径：零订单、零提交，并验证命令不再是 CLAIMED；
      • marker/report 不可读或证据不完整的负向路径：仍为 CLAIMED，零订单、零提交。

   当前无法安全给出具体补丁或测试结果：本次明确是 review-only，且快照没有提供 ExecutionCommandStore 的完整终态语义，无法确认哪个终态最准确，也不能声称测试已执行。ಿನ್ನೆ
4. Pigbibi commented on Sep 15, 2026

   

   Pigbibi

   on Sep 15, 2026

   CollaboratorAuthor

   More actions

   固定历史案例的依赖源码补充

   来源：LongBridgePlatform 26844fa9b909e98e867acf4c50120b90acc6c792 的 uv.lock 将 QuantPlatformKit 固定在 d51bb79dea8dc5773384d21b06e726620276558c。以下逐段复制该依赖提交的原始源码；不是当前 main，不含修复版本，不授予生产或交易权限。

   补充范围仅为持久命令的状态、认领与事件存储，以及既有 execution report 去重读取；写入范围仍仅限原批准的两个 LongBridge 文件。

   src/quant_platform_kit/common/execution_commands.py — ExecutionCommandState

   原始源码 L35–L44

   class ExecutionCommandState(str, Enum):
       QUEUED = "queued"
       CLAIMED = "claimed"
       SUBMITTED = "submitted"
       ACCEPTED = "accepted"
       PARTIALLY_FILLED = "partially_filled"
       FILLED = "filled"
       CANCELLED = "cancelled"
       REJECTED = "rejected"
       RECONCILIATION_REQUIRED = "reconciliation_required"

   src/quant_platform_kit/common/execution_commands.py — TERMINAL_EXECUTION_COMMAND_STATES

   原始源码 L47–L53

   TERMINAL_EXECUTION_COMMAND_STATES = frozenset(
       {
           ExecutionCommandState.FILLED,
           ExecutionCommandState.CANCELLED,
           ExecutionCommandState.REJECTED,
       }
   )

   src/quant_platform_kit/common/execution_commands.py — _ALLOWED_TRANSITIONS

   原始源码 L55–L104

   _ALLOWED_TRANSITIONS: dict[ExecutionCommandState, frozenset[ExecutionCommandState]] = {
       ExecutionCommandState.QUEUED: frozenset({ExecutionCommandState.CLAIMED, ExecutionCommandState.CANCELLED}),
       ExecutionCommandState.CLAIMED: frozenset(
           {
               ExecutionCommandState.SUBMITTED,
               ExecutionCommandState.RECONCILIATION_REQUIRED,
               ExecutionCommandState.CANCELLED,
               ExecutionCommandState.REJECTED,
           }
       ),
       ExecutionCommandState.SUBMITTED: frozenset(
           {
               ExecutionCommandState.ACCEPTED,
               ExecutionCommandState.PARTIALLY_FILLED,
               ExecutionCommandState.FILLED,
               ExecutionCommandState.CANCELLED,
               ExecutionCommandState.REJECTED,
               ExecutionCommandState.RECONCILIATION_REQUIRED,
           }
       ),
       ExecutionCommandState.ACCEPTED: frozenset(
           {
               ExecutionCommandState.PARTIALLY_FILLED,
               ExecutionCommandState.FILLED,
               ExecutionCommandState.CANCELLED,
               ExecutionCommandState.REJECTED,
               ExecutionCommandState.RECONCILIATION_REQUIRED,
           }
       ),
       ExecutionCommandState.PARTIALLY_FILLED: frozenset(
           {
               ExecutionCommandState.PARTIALLY_FILLED,
               ExecutionCommandState.FILLED,
               ExecutionCommandState.CANCELLED,
               ExecutionCommandState.RECONCILIATION_REQUIRED,
           }
       ),
       ExecutionCommandState.RECONCILIATION_REQUIRED: frozenset(
           {
               ExecutionCommandState.ACCEPTED,
               ExecutionCommandState.PARTIALLY_FILLED,
               ExecutionCommandState.FILLED,
               ExecutionCommandState.CANCELLED,
               ExecutionCommandState.REJECTED,
           }
       ),
       ExecutionCommandState.FILLED: frozenset(),
       ExecutionCommandState.CANCELLED: frozenset(),
       ExecutionCommandState.REJECTED: frozenset(),
   }

   src/quant_platform_kit/common/execution_commands.py — validate_execution_command_transition

   原始源码 L146–L153

   def validate_execution_command_transition(
       previous_state: ExecutionCommandState | str,
       next_state: ExecutionCommandState | str,
   ) -> None:
       previous = _normalize_state(previous_state)
       next_value = _normalize_state(next_state)
       if next_value not in _ALLOWED_TRANSITIONS[previous]:
           raise ValueError(f"invalid execution command transition: {previous.value} -> {next_value.value}")

   src/quant_platform_kit/common/execution_commands.py — is_due_on

   原始源码 L214–L215

       def is_due_on(self, as_of_date: object) -> bool:
           return _iso_date(as_of_date, field_name="as_of_date") == self.effective_date

   src/quant_platform_kit/common/execution_commands.py — ExecutionCommandStore

   原始源码 L323–L479

   @dataclass(frozen=True)
   class ExecutionCommandStore:
       """Create-only command and event storage with a fail-closed claim path."""
   
       local_dir: str | Path | None = None
       cloud_prefix_uri: str | None = None
       project_id: str | None = None
       namespace: str = DEFAULT_EXECUTION_COMMAND_NAMESPACE
       object_store: Any = None
   
       def enqueue(self, command: ExecutionCommand) -> bool:
           return self._create_text(
               self._command_location(command),
               json.dumps(command.to_dict(), ensure_ascii=False, indent=2, sort_keys=True),
           )
   
       def list_due(self, effective_date: object) -> tuple[ExecutionCommand, ...]:
           due_date = _iso_date(effective_date, field_name="effective_date")
           commands: list[ExecutionCommand] = []
           for location in self._list_locations(self._due_prefix(due_date)):
               if not str(location).endswith("/command.json") and Path(str(location)).name != "command.json":
                   continue
               commands.append(ExecutionCommand.from_dict(json.loads(self._read_text(location))))
           return tuple(sorted(commands, key=lambda command: command.command_id))
   
       def current_state(self, command: ExecutionCommand) -> ExecutionCommandState:
           events = self.events(command)
           return events[-1].state if events else ExecutionCommandState.QUEUED
   
       def events(self, command: ExecutionCommand) -> tuple[ExecutionCommandEvent, ...]:
           events: list[ExecutionCommandEvent] = []
           for location in self._list_locations(self._event_prefix(command)):
               if not str(location).endswith(".json"):
                   continue
               events.append(ExecutionCommandEvent.from_dict(json.loads(self._read_text(location))))
           events.sort(key=lambda event: event.sequence)
           expected_sequence = 1
           previous_state = ExecutionCommandState.QUEUED
           for event in events:
               if event.command_id != command.command_id or event.sequence != expected_sequence:
                   raise RuntimeError("execution command events are incomplete or belong to another command")
               if event.previous_state != previous_state:
                   raise RuntimeError("execution command event history has an invalid previous state")
               previous_state = event.state
               expected_sequence += 1
           return tuple(events)
   
       def claim_due(
           self,
           command: ExecutionCommand,
           *,
           as_of_date: object,
           claimant: object,
       ) -> ExecutionCommandEvent | None:
           if not command.is_due_on(as_of_date):
               return None
           return self.append_event(
               command,
               next_state=ExecutionCommandState.CLAIMED,
               details={"claimant": _safe_path_part(claimant, field_name="claimant")},
               expected_previous_state=ExecutionCommandState.QUEUED,
           )
   
       def append_event(
           self,
           command: ExecutionCommand,
           *,
           next_state: ExecutionCommandState | str,
           details: Mapping[str, Any] | None = None,
           expected_previous_state: ExecutionCommandState | str | None = None,
       ) -> ExecutionCommandEvent | None:
           prior_events = self.events(command)
           previous_state = prior_events[-1].state if prior_events else ExecutionCommandState.QUEUED
           if expected_previous_state is not None and previous_state is not _normalize_state(expected_previous_state):
               return None
           state = _normalize_state(next_state)
           validate_execution_command_transition(previous_state, state)
           sequence = len(prior_events) + 1
           event = ExecutionCommandEvent(
               command_id=command.command_id,
               sequence=sequence,
               previous_state=previous_state,
               state=state,
               recorded_at=datetime.now(timezone.utc).isoformat(),
               details_json=_canonical_json(dict(details or {})),
           )
           created = self._create_text(
               self._event_location(command, sequence),
               json.dumps(event.to_dict(), ensure_ascii=False, indent=2, sort_keys=True),
           )
           return event if created else None
   
       def _command_location(self, command: ExecutionCommand) -> str | Path:
           return self._location(command.effective_date, command.command_id, "command.json")
   
       def _event_prefix(self, command: ExecutionCommand) -> str | Path:
           return self._location(command.effective_date, command.command_id, "events")
   
       def _event_location(self, command: ExecutionCommand, sequence: int) -> str | Path:
           return self._location(command.effective_date, command.command_id, "events", f"{sequence:08d}.json")
   
       def _due_prefix(self, effective_date: str) -> str | Path:
           return self._location(effective_date)
   
       def _location(self, effective_date: str, *parts: str) -> str | Path:
           safe_date = _iso_date(effective_date, field_name="effective_date")
           safe_parts = [_safe_path_part(part, field_name="execution command path part") for part in parts]
           if self.cloud_prefix_uri:
               root = str(self.cloud_prefix_uri).rstrip("/")
               return "/".join((root, self.namespace, EXECUTION_COMMAND_SCHEMA_VERSION, safe_date, *safe_parts))
           if self.local_dir:
               return Path(self.local_dir) / self.namespace / EXECUTION_COMMAND_SCHEMA_VERSION / safe_date / Path(*safe_parts)
           raise RuntimeError("execution command store has no durable backend")
   
       def _create_text(self, location: str | Path, data: str) -> bool:
           if self.cloud_prefix_uri:
               store = self.object_store or self._object_store()
               create = getattr(store, "create_text", None)
               if not callable(create):
                   raise RuntimeError("cloud execution command store lacks atomic create-only support")
               return bool(create(str(location), data, "application/json"))
           path = Path(location)
           path.parent.mkdir(parents=True, exist_ok=True)
           temporary_path = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
           try:
               fd = os.open(temporary_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
               with os.fdopen(fd, "w", encoding="utf-8") as handle:
                   handle.write(data)
                   handle.flush()
                   os.fsync(handle.fileno())
               try:
                   os.link(temporary_path, path)
               except FileExistsError:
                   return False
               return True
           finally:
               temporary_path.unlink(missing_ok=True)
   
       def _read_text(self, location: str | Path) -> str:
           if self.cloud_prefix_uri:
               return (self.object_store or self._object_store()).read_text(str(location))
           return Path(location).read_text(encoding="utf-8")
   
       def _list_locations(self, prefix: str | Path) -> tuple[str | Path, ...]:
           if self.cloud_prefix_uri:
               return tuple((self.object_store or self._object_store()).list(str(prefix)))
           path = Path(prefix)
           if not path.exists():
               return ()
           return tuple(path.glob("*/command.json")) if path.name != "events" else tuple(path.glob("*.json"))
   
       def _object_store(self):
           try:
               from quant_platform_kit.cloud import get_object_store
           except ImportError as exc:
               raise RuntimeError("quant_platform_kit.cloud is required for cloud execution commands") from exc
           return get_object_store(project_id=self.project_id)

   src/quant_platform_kit/common/execution_state.py — has_prior_execution_report

   原始源码 L316–L368

       def has_prior_execution_report(
           self,
           *,
           platform: str,
           strategy_profile: str,
           account_scope: str,
           signal_date: object,
           effective_date: object,
           dry_run_only: bool,
       ) -> bool:
           if not self.cloud_prefix_uri:
               return False
           signal = _first_non_empty(signal_date)
           effective = _first_non_empty(effective_date)
           if not signal and not effective:
               return False
           month_segment = _month_segment(signal, effective)
           bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or ""))
           object_prefix = "/".join(
               part.strip("/")
               for part in (
                   prefix,
                   _runtime_report_segment(platform),
                   _runtime_report_segment(strategy_profile),
                   _runtime_report_segment(account_scope),
                   month_segment,
               )
               if part and part.strip("/")
           )
           store = self._object_store()
           prefix_uri = f"gs://{bucket_name}/{object_prefix}"
           scanned = 0
           for uri in store.list(prefix_uri):
               if not uri.endswith(".json"):
                   continue
               scanned += 1
               if scanned > max(1, int(self.prior_report_scan_limit or 1)):
                   break
               try:
                   payload = json.loads(store.read_text(uri))
               except Exception:
                   continue
               if _report_matches_execution(
                   payload,
                   platform=platform,
                   strategy_profile=strategy_profile,
                   account_scope=account_scope,
                   signal_date=signal,
                   effective_date=effective,
                   dry_run_only=dry_run_only,
               ):
                   return True
           return False

   src/quant_platform_kit/common/execution_state.py — _report_matches_execution

   原始源码 L638–L667

   def _report_matches_execution(
       payload: Mapping[str, Any],
       *,
       platform: str,
       strategy_profile: str,
       account_scope: str,
       signal_date: str,
       effective_date: str,
       dry_run_only: bool,
   ) -> bool:
       report = dict(payload or {})
       if _optional_str(report.get("platform")).lower() != _optional_str(platform).lower():
           return False
       if _optional_str(report.get("strategy_profile")).lower() != _optional_str(strategy_profile).lower():
           return False
       if _optional_str(report.get("account_scope")).lower() != _optional_str(account_scope).lower():
           return False
       if bool(report.get("dry_run")) != bool(dry_run_only):
           return False
       summary = dict(report.get("summary") or {})
       if signal_date and _date_key(signal_date) not in _report_signal_date_keys(report, summary):
           return False
       if effective_date and _date_key(effective_date) not in _report_effective_date_keys(report, summary):
           return False
       return (
           bool(summary.get("action_done"))
           or int(float(summary.get("orders_previewed_count") or 0)) > 0
           or int(float(summary.get("order_events_count") or 0)) > 0
           or _is_successful_no_action_report(report, summary)
       )

   src/quant_platform_kit/common/execution_state.py — _is_successful_no_action_report

   原始源码 L670–L675

   def _is_successful_no_action_report(report: Mapping[str, Any], summary: Mapping[str, Any]) -> bool:
       if _optional_str(report.get("status")).lower() != "ok":
           return False
       if int(float(summary.get("orders_skipped_count") or 0)) > 0:
           return False
       return bool("action_done" in summary and not summary.get("action_done"))

   src/quant_platform_kit/common/execution_state.py — _report_signal_date_keys

   原始源码 L678–L686

   def _report_signal_date_keys(report: Mapping[str, Any], summary: Mapping[str, Any]) -> set[str]:
       signal_snapshot = _report_signal_snapshot(report)
       return _date_keys(
           summary.get("signal_date"),
           signal_snapshot.get("signal_as_of"),
           signal_snapshot.get("market_date"),
           signal_snapshot.get("price_as_of"),
           signal_snapshot.get("snapshot_as_of"),
       )

   src/quant_platform_kit/common/execution_state.py — _report_effective_date_keys

   原始源码 L689–L694

   def _report_effective_date_keys(report: Mapping[str, Any], summary: Mapping[str, Any]) -> set[str]:
       signal_snapshot = _report_signal_snapshot(report)
       return _date_keys(
           summary.get("effective_date"),
           signal_snapshot.get("effective_date"),
       )

   src/quant_platform_kit/common/execution_state.py — _report_signal_snapshot

   原始源码 L697–L702

   def _report_signal_snapshot(report: Mapping[str, Any]) -> dict[str, Any]:
       diagnostics = report.get("diagnostics")
       if not isinstance(diagnostics, Mapping):
           return {}
       signal_snapshot = diagnostics.get("signal_snapshot")
       return dict(signal_snapshot) if isinstance(signal_snapshot, Mapping) else {}

   src/quant_platform_kit/common/execution_state.py — _date_keys

   原始源码 L705–L706

   def _date_keys(*values: object) -> set[str]:
       return {key for value in values if (key := _date_key(value))}

   src/quant_platform_kit/common/execution_state.py — _date_key

   原始源码 L709–L713

   def _date_key(value: object) -> str:
       text = _optional_str(value)
       if len(text) >= 10 and text[4] == "-" and text[7] == "-":
           return text[:10]
       return text
5. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   Codex Audit

   Codex execution failed with exit code 1.

   No files were pushed. Check the bridge workflow logs for details.
6. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   Codex Audit

   Codex execution failed with exit code 1.

   No files were pushed. Check the bridge workflow logs for details.
7. quantcrossrepoautomation commented on Sep 15, 2026

   

   quantcrossrepoautomationbot

   on Sep 15, 2026

   More actions

   Codex Audit

   修复摘要

   仅修改 application/rebalance_service.py 与 tests/test_rebalance_service.py。

   修复在已成功 claim 的 durable live command 上补充重复证据收敛：只有 marker/prior-report 读取成功、至少一项明确匹配、且当前周期未产生订单时，才以 expected_previous_state=CLAIMED 写入 CANCELLED。读取失败或事件写入返回 None 时保持 CLAIMED，继续 fail-closed。

   新增离线历史回归覆盖：

   • 2026-09-10 signal → 2026-09-11 effective，marker 缺失但 prior report 明确存在，验证一次 claim、零订单、最终 CANCELLED；
   • prior-report 读取失败，验证实际读取、无 close event、状态保持 CLAIMED、零订单。

   未执行本地测试；按约定由 AIAuditBridge 应用补丁并运行 pinned bounded test。

   Created fix PR: #483

   This PR is high-risk for unattended merge and requires human review.

   • Risk level: high
   • Changed lines: 108
   • Reasons:
     • invalid auto-merge policy schema requires human review
   • High-risk files:
     • application/rebalance_service.py
     • tests/test_rebalance_service.py

   Added human-review-required to the PR so operators can review it before merge.