[Historical AI validation] SG duplicate execution leaves claimed command unresolved

[Pigbibi]

Historical SG rebalance claim deduplication bug

• Repository: QuantStrategyLab/LongBridgePlatform
• Source ref: ai-history-sg-claim-dedup-20260915
• Baseline revision: 26844fa9b909e98e867acf4c50120b90acc6c792
• Execution mode: historical offline sandbox only

Problem

In the SG rebalance path, a due durable command is claimed before execution. In
the same session, the service can then find evidence in the execution marker or
prior report that the signal was already executed. The service skips duplicate
execution, but the current claim remains in CLAIMED, so later processing sees a
stale in-flight command and is blocked.

The historical evidence is:

• The affected cycle is SG 9/11, with signal 9/10 and effective date 9/11.
• The command was claimed once (cmd-3915a5fb16264cb4eab5cd4a970de069).
• Same-cycle evidence says the duplicate was skipped because the signal dated 9/10
  for the 9/11 execution session had already been recorded.
• The observed report had action_done=false,
  broker_submission_done=false, and orders=0.
• The later unresolved state is caused by the claimed command remaining
  durable, not by evidence of a broker submission.

Requested repair

Inspect the existing claim, marker, prior-report, and terminal-state flow and
propose the smallest repair that closes the current claim when duplicate
execution is positively confirmed and no new submission is made. Preserve the
existing duplicate-execution protection.

Do not cancel or otherwise terminally resolve a claim when marker/report
reading fails, identity cannot be established, evidence is ambiguous, or the
broker outcome is unknown. Those cases must remain fail-closed and retain the
claim for reconciliation.

The repair and regression coverage may modify only:

• application/rebalance_service.py
• tests/test_rebalance_service.py

The regression must be deterministic, offline, and use historical fixtures. It
must cover both the positively confirmed duplicate path and a negative path in
which marker/report evidence is unavailable or unreadable; the latter must
remain CLAIMED with zero orders/submissions. Do not perform live orders,
deployments, remote state changes, or workflow dispatch.

Required response

Return the proposed patch and tests for human review, with the exact files
changed, the evidence used, and the test command/result. If the source or
evidence needed to establish a safe terminal state is absent, stop and report
the gap instead of guessing.

This is an authorized historical research task, not an active production incident. The bridge may create one draft PR against the historical sandbox branch after its isolated tests pass. Do not merge, deploy, or modify the production main branch.

---

Previous attempt diagnostic feedback (append-only; preserve the issue body above)

Use the fixed repository state only: QuantStrategyLab/LongBridgePlatform, branch ai-history-sg-claim-dedup-20260915, baseline commit 26844fa9b909e98e867acf4c50120b90acc6c792. The allowed write set remains exactly application/rebalance_service.py and tests/test_rebalance_service.py; do not modify other files, prompts, frameworks, grants, claims, deployment, or trading behavior.

The prior generated test was diagnosed offline with these concrete failures: CommandStore.current_state was always CLAIMED; enqueue returned the same already-due command, so the positive path had duplicate due work; the frozen plan dates were inconsistent; and the append_event return value of None was not verified. These are diagnostic facts, not a request to copy the previous human repair.

Regenerate and implement the smallest fix in the two allowed files against the supplied baseline. The positive test must exercise the real path from an initially QUEUED command to one claim, use an independent future command for the new signal, and use the frozen plan date transition 2026-09-10 to 2026-09-11 with the correct marker key. It must verify the actual marker/prior-report reads, final state CANCELLED, exactly one claim, and zero orders.

Keep negative cases independent and verify they actually read the evidence, remain CLAIMED, emit no close event, and submit zero orders. Cover marker-missing/prior-report-present behavior. If evidence reads fail, or append_event returns None, do not claim closure; preserve the unresolved/claimed state and emit the existing warning behavior without inventing a new return protocol.

Return targeted edits against the supplied baseline. The bridge, not this model session, applies the patch and runs the pinned offline tests; do not claim test execution or results yourself. Do not claim AI success from a human patch, do not call another model, do not dispatch production work, and do not merge or publish.

[quantcrossrepoautomation]

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

[quantcrossrepoautomation]

Codex Audit

Codex execution failed with exit code 1.

No files were pushed. Check the bridge workflow logs for details.

[quantcrossrepoautomation]

结论：问题已在提供的快照中得到确认，属于“已成功认领但重复证据分支未收敛命令状态”的状态机缺陷。它不会直接证明重复下单，但会让后续周期持续被旧的 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 的完整终态语义，无法确认哪个终态最准确，也不能声称测试已执行。ಿನ್ನೆ

[Pigbibi]

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

来源：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

[quantcrossrepoautomation]

Codex Audit

Codex execution failed with exit code 1.

No files were pushed. Check the bridge workflow logs for details.

[quantcrossrepoautomation]

Codex Audit

Codex execution failed with exit code 1.

No files were pushed. Check the bridge workflow logs for details.

[quantcrossrepoautomation]

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.