"""Shared durable boundary for non-asset Feishu approval creation. The provider UUID and immutable create payload are committed before the outbound POST. A retry reuses that UUID; Feishu's 60012 collision is resolved through the bounded list/detail lookup implemented by ``FeishuApprovalClient``. Domain services remain responsible for atomically materialising their own row, the parallel ``ApprovalRequest``, audit evidence, and the intent's ``completed`` transition after this module returns ``provider_created``. """ from __future__ import annotations import hashlib import json from collections.abc import Iterator from contextlib import contextmanager from contextvars import ContextVar from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID, uuid4 import httpx from fastapi.encoders import jsonable_encoder from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.feishu_client import parse_feishu_error_code from app.approval.models import ApprovalSubmissionIntent from app.audit.service import AuditService from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger, sanitize_log_text from app.integrations.feishu_widget_values import ( FEISHU_FORM_VALUE_ERROR_CODES, form_without_date_interval, rejection_targets_date_interval, ) logger = get_safe_logger(__name__) _RECOVERABLE_STATES = ( "prepared", "submitting", "unknown", "provider_created", "compensation_pending", ) _TERMINAL_STATES = frozenset({"completed", "compensated", "failed"}) _MAX_RETRY_DELAY_MINUTES = 60 _PROVIDER_ATTEMPT_ANCHORS = "provider_attempt_anchors" _MATERIALIZED_APPROVAL_REQUEST_IDS: ContextVar[set[UUID] | None] = ContextVar( "materialized_approval_request_ids", default=None, ) @contextmanager def collect_materialized_approval_request_ids() -> Iterator[set[UUID]]: """Collect non-asset approval rows finalized by this async task. The drift handler uses this narrow, task-local marker around its intent recovery phase. It therefore sees rows materialized by that exact run, without widening the normal age-gated scan to unrelated young requests created concurrently by HTTP workers. """ collected: set[UUID] = set() token = _MATERIALIZED_APPROVAL_REQUEST_IDS.set(collected) try: yield collected finally: _MATERIALIZED_APPROVAL_REQUEST_IDS.reset(token) def compact_terminal_intent_payload( intent: ApprovalSubmissionIntent, *, resource_type: str, resource_id: str | UUID, ) -> None: """Replace recovery-only data with the minimal terminal replay receipt. Provider payloads, form answers, request metadata, IP addresses and user agents are needed only until recovery/finalization succeeds. Stable idempotency and provider correlations live in dedicated columns, while the small resource pointer below lets terminal retries return the domain row. """ if intent.state not in _TERMINAL_STATES: raise ValueError("submission intent must be terminal before compaction") collector = _MATERIALIZED_APPROVAL_REQUEST_IDS.get() if ( collector is not None and intent.scope != "asset_request" and intent.approval_request_id is not None ): collector.add(intent.approval_request_id) intent.payload = { "schema": "terminal-v1", "resource_type": resource_type, "resource_id": str(resource_id), } intent.next_attempt_at = None def terminal_intent_resource_id( intent: ApprovalSubmissionIntent, *, resource_type: str, ) -> str: """Read and validate the minimal resource pointer used by terminal replay.""" payload = dict(intent.payload or {}) if ( payload.get("schema") != "terminal-v1" or payload.get("resource_type") != resource_type or not payload.get("resource_id") ): raise RuntimeError("terminal submission intent has no valid resource receipt") return str(payload["resource_id"]) def reconciliation_due_predicate(now: datetime) -> Any: """SQL predicate shared by every domain's bounded reconciliation batch.""" return (ApprovalSubmissionIntent.next_attempt_at.is_(None)) | ( ApprovalSubmissionIntent.next_attempt_at <= now ) def reconciliation_order() -> tuple[Any, ...]: """Deterministic fair order after deferred rows become eligible again.""" return ( func.coalesce( ApprovalSubmissionIntent.next_attempt_at, ApprovalSubmissionIntent.created_at, ).asc(), ApprovalSubmissionIntent.created_at.asc(), ApprovalSubmissionIntent.id.asc(), ) async def defer_submission_intent_retry( db: AsyncSession, intent_id: UUID, exc: BaseException, ) -> None: """Persist bounded backoff so one poison row cannot starve a fixed batch.""" await db.rollback() intent = await db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None or intent.state not in _RECOVERABLE_STATES: await db.rollback() return schedule_submission_intent_retry(intent, exc) await db.commit() def schedule_submission_intent_retry( intent: ApprovalSubmissionIntent, exc: BaseException, ) -> None: """Stage bounded retry scheduling in the caller's current transaction.""" if intent.state not in _RECOVERABLE_STATES: raise ValueError("cannot schedule a terminal submission intent") intent.attempt_count += 1 delay_minutes = min( 2 ** min(max(intent.attempt_count - 1, 0), 6), _MAX_RETRY_DELAY_MINUTES, ) intent.next_attempt_at = datetime.now(UTC) + timedelta(minutes=delay_minutes) intent.last_error = sanitize_log_text(exc, max_len=512) def submission_fingerprint(payload: dict[str, Any]) -> str: """Hash caller semantics without persisting an additional plaintext key.""" canonical = json.dumps( jsonable_encoder(payload), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def record_provider_attempt_anchor( intent: ApprovalSubmissionIntent, *, attempted_at: datetime | None = None, ) -> datetime: """Persist the time window for a provider create before outbound I/O.""" anchor = attempted_at or datetime.now(UTC) if anchor.tzinfo is None: anchor = anchor.replace(tzinfo=UTC) else: anchor = anchor.astimezone(UTC) payload = dict(intent.payload or {}) anchors = list(payload.get(_PROVIDER_ATTEMPT_ANCHORS) or []) encoded = anchor.isoformat() if encoded not in anchors: anchors.append(encoded) payload[_PROVIDER_ATTEMPT_ANCHORS] = anchors intent.payload = payload return anchor def provider_attempt_anchors( payload: dict[str, Any], *, fallback: datetime, ) -> tuple[datetime, ...]: """Return newest-first durable lookup anchors, with legacy fallback.""" parsed: list[datetime] = [] for raw in payload.get(_PROVIDER_ATTEMPT_ANCHORS) or []: try: anchor = datetime.fromisoformat(str(raw)) except (TypeError, ValueError): continue if anchor.tzinfo is None: anchor = anchor.replace(tzinfo=UTC) else: anchor = anchor.astimezone(UTC) if anchor not in parsed: parsed.append(anchor) if not parsed: anchor = fallback if anchor.tzinfo is None: anchor = anchor.replace(tzinfo=UTC) else: anchor = anchor.astimezone(UTC) parsed.append(anchor) return tuple(reversed(parsed)) async def submit_with_durable_intent( db: AsyncSession, *, client: Any, scope: str, applicant_id: UUID, semantic_payload: dict[str, Any], provider_payload: dict[str, Any], finalization_payload: dict[str, Any], trace_id: str, recovery_key: str | None = None, intent_key: str | None = None, ) -> ApprovalSubmissionIntent: """Return an intent durably correlated to one Feishu instance. Callers must hold their existing logical-submission lock. Within that lock, the latest unfinished intent for ``(scope, applicant)`` is the retry anchor. ``recovery_key`` partitions unfinished implicit-key submissions for domains that permit several logical requests per applicant. Domains that already own a durable request identity pass ``intent_key`` and recover that exact generation instead. A different payload is rejected while an unknown create is being reconciled; completed/failed history does not prevent a later legitimate request in the implicit-key mode. """ if intent_key is not None and (not intent_key or len(intent_key) > 128): raise ValueError("intent_key must contain 1..128 characters") fingerprint = submission_fingerprint(semantic_payload) intent_predicates = [ ApprovalSubmissionIntent.scope == scope, ApprovalSubmissionIntent.applicant_id == applicant_id, ] if intent_key is None: intent_predicates.append( ApprovalSubmissionIntent.state.in_(_RECOVERABLE_STATES), ) else: intent_predicates.append( ApprovalSubmissionIntent.idempotency_key == intent_key, ) intent_query = ( select(ApprovalSubmissionIntent) .where(*intent_predicates) .order_by(ApprovalSubmissionIntent.created_at.desc()) .with_for_update() .execution_options(populate_existing=True) ) if intent_key is None: candidates = list((await db.scalars(intent_query)).all()) intent = next( ( candidate for candidate in candidates if (candidate.payload or {}).get("recovery_key") == recovery_key ), None, ) else: intent = await db.scalar(intent_query.limit(1)) if intent is not None and intent.payload_fingerprint != fingerprint: raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message=( "A previous approval submission is still being reconciled. " "Retry the original request before submitting different data." ), status_code=409, ) if intent is None: intent = ApprovalSubmissionIntent( scope=scope, applicant_id=applicant_id, # These domains already serialize one pending submission per # requester. The random key identifies this generation; retries # locate the unfinished row by scope/applicant and never mint a # second provider UUID. idempotency_key=intent_key or str(uuid4()), payload_fingerprint=fingerprint, provider_request_uuid=str(uuid4()), state="prepared", payload=jsonable_encoder( { "semantic": semantic_payload, "provider": provider_payload, "finalization": finalization_payload, "trace_id": trace_id, "recovery_key": recovery_key, } ), ) db.add(intent) await db.flush() if intent.state in {"provider_created", "compensation_pending", "completed"}: return intent if intent.state == "failed": raise AppException( code="APPROVAL_SUBMISSION_PREVIOUSLY_FAILED", message=( "This durable approval submission was rejected before " "provider acceptance. Start a new logical request." ), status_code=409, ) intent.state = "submitting" intent.attempt_count += 1 intent.last_error = None intent.next_attempt_at = None intent_id = intent.id provider_uuid = intent.provider_request_uuid stored_provider = dict((intent.payload or {}).get("provider") or {}) await db.commit() # durable before provider I/O identity_type = stored_provider["identity_type"] if identity_type not in {"user_id", "open_id"}: raise RuntimeError("invalid persisted Feishu applicant identity type") async def _create(form: str) -> dict[str, Any]: # ``client`` is duck-typed (Any) so the domains can inject test # doubles; pin the return locally to keep the mypy ratchet flat. created: dict[str, Any] = await client.create_instance( approval_code=stored_provider["approval_code"], user_id=(stored_provider["identity_value"] if identity_type == "user_id" else None), open_id=(stored_provider["identity_value"] if identity_type == "open_id" else None), form=form, node_approver_user_id_list=stored_provider.get("node_approver_user_id_list"), node_approver_open_id_list=stored_provider.get("node_approver_open_id_list"), node_cc_user_id_list=stored_provider.get("node_cc_user_id_list"), node_cc_open_id_list=stored_provider.get("node_cc_open_id_list"), instance_uuid=provider_uuid, ) return created try: try: instance = await _create(stored_provider["form"]) except AppException as exc: # Borrow / validity window degradation (2026-08-14). The # ``dateInterval`` value shape these flows ship is read off # live Feishu, but its accepted WRITE shape could not be # exercised end-to-end without submitting into a production # approval flow. If Feishu rejects the form value, resubmit # once without the interval widget — the shape that shipped # between 2026-05-19 and 2026-08-14, which Feishu is known # to accept. Worst case: the approver's card is missing the # dates (the old behaviour), never "nobody can file this # request". The 2026-08-04 amount-widget regression had no # such rail and blocked every non-CNY buyback for 9 days. # # A rejected create leaves nothing on Feishu's side, so the # retry safely reuses ``provider_uuid``. Every other # exception — 60012 collisions included — falls through to # the handlers below untouched. if parse_feishu_error_code(exc) not in FEISHU_FORM_VALUE_ERROR_CODES: raise # Only degrade when the rejection actually implicates the # range. Feishu names the offending widget (``控件= ``), # so a rejection caused by some other widget stays terminal # instead of burning a retry that would fail identically. if not rejection_targets_date_interval( stored_provider["form"], exc.message or "", ): raise degraded_form = form_without_date_interval(stored_provider["form"]) if degraded_form is None: raise logger.warning( "Feishu rejected the dateInterval value for approval %s " "(scope=%s): %s — resubmitting without the date range. " "The approver's card will not show the dates.", stored_provider.get("approval_code"), scope, exc.message, ) stored_provider["form"] = degraded_form # Durable before the retry, same contract as the commit # above: whatever Feishu accepts is what a later recovery # tick must replay. await _persist_degraded_form(db, intent_id, degraded_form) instance = await _create(degraded_form) except AppException as exc: if parse_feishu_error_code(exc) != 60012: await _record_failure(db, intent_id, exc, unknown=False) raise intent = await db.get(ApprovalSubmissionIntent, intent_id) if intent is None: raise RuntimeError("approval submission intent disappeared") from exc anchor = intent.created_at if anchor.tzinfo is None: anchor = anchor.replace(tzinfo=UTC) try: instance = await client.find_instance_by_uuid( approval_code=stored_provider["approval_code"], instance_uuid=provider_uuid, start_time=anchor - timedelta(minutes=5), end_time=min( anchor + timedelta(minutes=5), datetime.now(UTC) + timedelta(minutes=1), ), ) except Exception as lookup_exc: await _record_failure(db, intent_id, lookup_exc, unknown=True) raise _unknown_result_error() from lookup_exc if instance is None: lookup_miss = RuntimeError( "Feishu UUID collision was not visible in the bounded window" ) await _record_failure(db, intent_id, lookup_miss, unknown=True) raise _unknown_result_error() from lookup_miss except httpx.TransportError as exc: await _record_failure(db, intent_id, exc, unknown=True) raise _unknown_result_error() from exc except Exception as exc: # The exception surfaced from the create-stage call, but its type does # not prove that the POST was never sent. Response decoding and other # client-side post-processing can fail after Feishu accepted the UUID. # Preserve the generation and force same-UUID reconciliation instead # of declaring a false terminal failure that lets implicit-key callers # mint a second provider instance. await _record_failure(db, intent_id, exc, unknown=True) raise _unknown_result_error() from exc external_id = str((instance or {}).get("instance_code") or "").strip() if not external_id: missing_code = RuntimeError("Feishu create response omitted instance_code") await _record_failure(db, intent_id, missing_code, unknown=True) raise _unknown_result_error() from missing_code await db.rollback() intent = await db.get(ApprovalSubmissionIntent, intent_id) if intent is None: raise RuntimeError("approval submission intent disappeared") payload = dict(intent.payload or {}) payload["provider_response"] = jsonable_encoder(instance) intent.payload = payload intent.external_id = external_id intent.state = "provider_created" intent.last_error = None intent.next_attempt_at = None await db.commit() return intent async def _persist_degraded_form( db: AsyncSession, intent_id: UUID, degraded_form: str, ) -> None: """Persist the form Feishu will actually be asked to accept. Called between the rejected create and the retry, so a later recovery tick replays the degraded form rather than the one Feishu already refused. A missing intent row here is not fatal — the retry still goes out with the degraded form in memory. """ intent = await db.get(ApprovalSubmissionIntent, intent_id) if intent is None: return payload = dict(intent.payload or {}) provider = dict(payload.get("provider") or {}) provider["form"] = degraded_form provider["date_interval_dropped"] = True payload["provider"] = provider intent.payload = payload await db.commit() async def _record_failure( db: AsyncSession, intent_id: UUID, exc: BaseException, *, unknown: bool, ) -> None: await db.rollback() intent = await db.get(ApprovalSubmissionIntent, intent_id) if intent is None: return intent.state = "unknown" if unknown else "failed" intent.last_error = sanitize_log_text(exc, max_len=512) if unknown: delay_minutes = min( 2 ** min(max(intent.attempt_count - 1, 0), 6), _MAX_RETRY_DELAY_MINUTES, ) intent.next_attempt_at = datetime.now(UTC) + timedelta(minutes=delay_minutes) else: intent.next_attempt_at = None payload = dict(intent.payload or {}) await AuditService(db).log( trace_id=str(payload.get("trace_id") or f"approval-intent:{intent_id}"), actor_type="user", actor_user_id=intent.applicant_id, action=( "approval.submission_intent.unknown" if unknown else "approval.submission_intent.failed" ), resource_type="approval_submission_intent", resource_id=str(intent_id), status="failure", detail={ "scope": intent.scope, "provider_request_uuid": intent.provider_request_uuid, "state": intent.state, "error_type": type(exc).__name__, }, ) if not unknown: compact_terminal_intent_payload( intent, resource_type="approval_submission", resource_id=intent.id, ) await db.commit() def _unknown_result_error() -> AppException: return AppException( code="APPROVAL_SUBMISSION_RESULT_UNKNOWN", message=( "Feishu may have accepted this approval. Helix retained the " "submission identity; retry the same request to reconcile it." ), status_code=503, )