"""Asset inventory (盘点) service. Owns the lifecycle of an inventory cycle: * ``create_cycle`` — admin (or scheduled job) creates a draft cycle. * ``start_cycle`` — snapshot eligible assets, route each to a recipient bucket (assignee / location_admin / it_fallback), persist ``asset_inventory_items`` rows, dispatch one Feishu card per recipient, flip cycle to ``active``. * ``respond_to_item`` — recipient (or IT, on behalf) records a per-item response. Validates response_type against holder_kind and flips cycle counters + denormalised columns on ``asset_items`` for confirmed/issue items. * ``send_cycle_reminder`` — admin button OR scheduled job; one-shot re-push to recipients with pending items, idempotent on ``cycle.reminder_sent_at`` and per-item ``reminded_at``. * ``close_cycle`` — admin closes; pending items are frozen as-is and do NOT update ``last_inventory_at``. * ``preview_cycle`` — dry-run snapshot for a draft, returning recipient breakdown without persisting items. Cross-domain reads: * ``identity_persons`` — recipient name + active filter. * ``asset_office_addresses`` — idle-asset routing. These are **read-only** queries on tables the asset domain already owns (``asset_office_addresses``) or treats as identity infra (``identity_persons``). Per PLATFORM.md's refined rule, this is within the asset domain's data layer. Notification dispatch goes through ``app.notifications.NotificationService`` — a platform service, not a peer domain — so the no-cross-domain-import rule isn't violated. """ from __future__ import annotations from collections import defaultdict from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import asc, case, desc, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ( ConflictError, ForbiddenError, NotFoundError, ValidationError, ) from app.core.safe_logging import get_safe_logger from app.core.working_hours import ( is_local_working_time, local_start_of_day, resolve_timezone, ) from app.domains.asset.inventory_schemas import ( ALLOWED_CYCLE_STATUSES, ALLOWED_HOLDER_KINDS, ALLOWED_ITEM_STATUSES, CONFIRMING_RESPONSES, EXCLUDED_ASSET_STATUSES, HOLDER_KIND_LOCATION_ADMIN, HOLDER_KINDS_LOCATION_ADMIN, RESPONSE_TYPES_BY_HOLDER_KIND, CycleScopeFilter, sanitize_evidence_photo_keys, ) from app.domains.asset.location_resolver import ( LocationIndex, ManagedLocationResolver, ) from app.domains.asset.models import ( AssetHistory, AssetInventoryCycle, AssetInventoryItem, AssetItem, AssetLocation, AssetOfficeAddress, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson from app.notifications.models import NotificationOutbox from app.notifications.service import ( NotificationEnvelope, NotificationService, ) logger = get_safe_logger(__name__) # Template key for the optional manager / HRBP escalation digest sent # when an admin ticks "也通知直线 Lead / HRBP" on a manual remind. One # aggregated card per Lead / HRBP — never one card per pending employee # — so a manager with 10 unresponsive reports gets ONE nudge, not ten # (anti-storm, see [[feedback_no_new_pushes_to_employees]]). _ESCALATION_TEMPLATE_KEY = "asset_inventory_escalation" @dataclass(slots=True) class ReminderOutcome: """Result of one ``send_cycle_reminder`` call. ``recipients_notified`` — distinct holders pinged about their own pending items (the always-on behaviour). ``managers_notified`` / ``hrbps_notified`` — distinct Leads / HRBPs sent an aggregated nudge digest, both 0 unless the admin opted in on this call. """ recipients_notified: int = 0 managers_notified: int = 0 hrbps_notified: int = 0 from app.notifications import cadence # noqa: E402 _NOTIFICATION_TEMPLATE_KEY = "asset_inventory_notify" _REMINDER_TEMPLATE_KEY = "asset_inventory_reminder" # Days to wait after ``cycle.started_at`` before the first daily # reminder fires. One workday of grace — the launch notification # already went out, so the next-day nag is the first reminder. _REMINDER_GRACE_DAYS = 1 # Fence freshly-enqueued outbox rows out of the hourly # ``notification.outbox_retry`` picker while THIS process is still # draining delivery itself. The picker selects PENDING rows with # ``next_attempt_at IS NULL OR <= now`` (NULLs first!), so a phase-1 # row committed with ``next_attempt_at=NULL`` is claimed at the very # next hourly tick — mid-drain, that means the worker AND phase 2 both # send the same card (duplicate Feishu DM). Setting the fence keeps # the row invisible to the picker for the drain window; if the drain # process dies, the worker picks the row up once the fence expires — # self-healing, no loss. A row phase 2 successfully delivers flips to # DELIVERED (``next_attempt_at=NULL``), which the picker's status # filter excludes, so post-fence retry never double-sends it. _OUTBOX_RETRY_FENCE = timedelta(minutes=30) # ``force=True`` reminder calls bypass the per-local-day dedup but NOT # the up-front claim: a ``reminded_at`` stamp younger than this grace # means another caller (cron tick, double-clicked 立即催办) claimed the # recipient moments ago and may still be mid-send — re-claiming would # double-send. Sized to comfortably cover a company-wide fan-out. _FORCE_RECLAIM_GRACE = timedelta(minutes=10) # Minutes to wait after ``cycle.started_at`` before the launch # reconciler treats a holder with no outbox row as LOST rather than # still-being-enqueued by the in-flight background drain. Phase 1 of # the fan-out (all-DB, per-row commit) finishes in seconds even for a # company-wide cycle, so 30 minutes cannot race a healthy drain. _LAUNCH_RECONCILE_GRACE_MINUTES = 30 def _bucket_key(holder_kind: str) -> str: """Canonical fan-out bucket for a stored ``holder_kind``. The reminder buckets below are indexed **directly** by this value, so a token with no matching key raises KeyError in the middle of a fan-out. Rows written before migration 543 carry ``office_admin``, and the deploy applies migrations before it restarts containers, so an old container can still write one against a migrated table — both tokens therefore have to land in the same bucket. """ if holder_kind in HOLDER_KINDS_LOCATION_ADMIN: return HOLDER_KIND_LOCATION_ADMIN return holder_kind class AssetInventoryService: def __init__(self, db: AsyncSession) -> None: self.db = db # ── Cycle CRUD ──────────────────────────────────────────────────── async def list_cycles( self, *, status: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[AssetInventoryCycle], int]: stmt = select(AssetInventoryCycle).order_by( desc(AssetInventoryCycle.created_at), ) count_stmt = select(func.count()).select_from(AssetInventoryCycle) if status: if status not in ALLOWED_CYCLE_STATUSES: raise ValidationError(f"unknown status: {status}") stmt = stmt.where(AssetInventoryCycle.status == status) count_stmt = count_stmt.where(AssetInventoryCycle.status == status) total = (await self.db.execute(count_stmt)).scalar() or 0 offset = max(0, (page - 1) * page_size) stmt = stmt.offset(offset).limit(page_size) rows = (await self.db.execute(stmt)).scalars().all() return list(rows), total async def get_cycle(self, cycle_id: UUID) -> AssetInventoryCycle: cycle = await self.db.get(AssetInventoryCycle, cycle_id) if cycle is None: raise NotFoundError(f"inventory cycle {cycle_id} not found") return cycle async def create_cycle( self, *, name: str, description: str | None, scope_filter: CycleScopeFilter | None, due_at: datetime | None, created_by: UUID | None, ) -> AssetInventoryCycle: cycle = AssetInventoryCycle( name=name.strip(), description=(description or None), status="draft", # ``mode='json'`` coerces UUID → str (and datetime → ISO, # etc.) so the resulting dict is round-trippable through the # JSON column. Without it ``person_ids: list[UUID]`` reaches # SQLAlchemy's json_serializer with raw ``UUID`` objects and # ``json.dumps`` raises ``TypeError: Object of type UUID is # not JSON serializable`` — the first cycle with a non-empty # ``person_ids`` 500s before any row is written. Bug was # latent until the create-cycle form started exposing the # person picker; pre-PR cycles never carried UUIDs in # scope_filter. scope_filter=( scope_filter.model_dump(exclude_none=True, mode="json") if scope_filter is not None else {} ), due_at=due_at, created_by=created_by, ) self.db.add(cycle) await self.db.flush() return cycle async def update_cycle( self, cycle_id: UUID, *, name: str | None = None, description: str | None = None, due_at: datetime | None = None, clear_due_at: bool = False, ) -> AssetInventoryCycle: cycle = await self.get_cycle(cycle_id) if cycle.status == "closed": raise ConflictError("closed cycles cannot be edited") if name is not None: cycle.name = name.strip() if description is not None: cycle.description = description or None # ``due_at=None`` is ambiguous on a PATCH: an untouched field and # an explicitly-cleared DatePicker both arrive as None. The router # disambiguates via ``model_fields_set`` and passes # ``clear_due_at`` — without it, clearing the deadline in the edit # modal would report success and silently keep the old date. if clear_due_at: cycle.due_at = None elif due_at is not None: cycle.due_at = due_at await self.db.flush() return cycle async def delete_draft(self, cycle_id: UUID) -> None: cycle = await self.get_cycle(cycle_id) if cycle.status != "draft": raise ConflictError( "only draft cycles can be deleted; close active ones instead", ) await self.db.delete(cycle) await self.db.flush() # ── Snapshot + start ────────────────────────────────────────────── async def _query_eligible_assets( self, scope: dict[str, Any] | None, ) -> list[AssetItem]: """All assets in scope and not in an excluded status.""" # Soft-deleted assets (migration 115) keep their pre-delete # ``status`` (``delete_asset`` stamps ``deleted_at`` but never # touches ``status``), so filtering on status alone lets a # deleted-but-still-``available``/``idle`` row leak into the # cycle snapshot as a phantom count line no holder can # reconcile. Exclude soft-deleted rows explicitly. stmt = select(AssetItem).where( AssetItem.status.notin_(EXCLUDED_ASSET_STATUSES), AssetItem.deleted_at.is_(None), ) if scope: assignment_scope = scope.get("assignment_scope") or "all" if assignment_scope == "assigned": # Only rows currently held by someone — leaver assets # included (the classify step still routes them to # location_admin / it_fallback). stmt = stmt.where(AssetItem.assigned_to.is_not(None)) elif assignment_scope == "idle": # Idle / warehouse stock only — every row will route # to location_admin or it_fallback at classify time. stmt = stmt.where(AssetItem.assigned_to.is_(None)) # ``all`` is the no-op default. departments = scope.get("departments") or [] if departments: # AssetItem has no ``department`` column — it's derived # from the assignee's primary_department at read time # (see ``service.enrich_with_users``). Filtering by # department therefore only narrows the *assignee* # bucket: an idle laptop in 上海 has no department to # match, so it's excluded when this filter is set. # ``primary_department`` is a Python property, not a # column — derived from the primary org_membership's # ``org_node_name``. Translate to a SQL subquery # joining ``identity_org_memberships`` on # is_primary + active membership status. dept_persons = ( select(IdentityOrgMembership.person_id).where( IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), IdentityOrgMembership.org_node_name.in_(departments), ) ) stmt = stmt.where(AssetItem.assigned_to.in_(dept_persons)) locations = scope.get("locations") or [] if locations: stmt = stmt.where(AssetItem.location.in_(locations)) categories = scope.get("categories") or [] if categories: stmt = stmt.where(AssetItem.category.in_(categories)) person_ids = scope.get("person_ids") or [] if person_ids: # For person-scoped cycles, only the assignee bucket is # meaningful — idle assets won't surface. stmt = stmt.where(AssetItem.assigned_to.in_(person_ids)) result = await self.db.execute(stmt) return list(result.scalars().all()) async def _resolve_location_admins( self, ) -> tuple[dict[UUID, UUID], dict[UUID, UUID], LocationIndex]: """Build the routing lookups for the classifier. Returns ``(primary_admin_by_location, location_by_office, index)``: - ``primary_admin_by_location`` — the one person a location's idle assets are addressed to. The *primary* administrator specifically, not every administrator: a confirmation request is an "you are accountable for these" message, and fanning it to a location's whole admin list would multiply the volume while diffusing the responsibility it exists to place. :func:`resolve_location_primary_admins_bulk` falls back to the sole administrator when no primary flag is set, and yields nothing when a location has several and no primary — those assets go to the IT bucket rather than to an arbitrary pick. - ``location_by_office`` — for office-held assets, which managed location the holding office sits in. - ``index`` — the shared resolver, so the classifier can turn an asset's free-text location into a location id with exactly the same rules the region write-gate uses. An asset a regional admin is allowed to manage is therefore also routed to them at inventory start. """ from app.domains.asset.location_admin_sync import ( resolve_location_primary_admins_bulk, ) from app.domains.asset.location_resolver import LocationIndex from app.domains.asset.office_local_admins import ( resolve_office_location_ids_bulk, ) index = await LocationIndex.load(self.db) location_ids = [ row[0] for row in ( await self.db.execute(select(AssetLocation.id)) ).all() ] primaries = await resolve_location_primary_admins_bulk( self.db, location_ids, ) primary_admin_by_location: dict[UUID, UUID] = { location_id: person_id for location_id, person_id in primaries.items() if person_id is not None } office_ids = [ row[0] for row in ( await self.db.execute( select(AssetOfficeAddress.id).where( AssetOfficeAddress.is_active.is_(True), ) ) ).all() ] office_locations = await resolve_office_location_ids_bulk( self.db, office_ids, index=index, ) location_by_office: dict[UUID, UUID] = { office_id: location_id for office_id, location_id in office_locations.items() if location_id is not None } return primary_admin_by_location, location_by_office, index async def _active_person_ids( self, person_ids: set[UUID], ) -> set[UUID]: if not person_ids: return set() rows = ( await self.db.execute( select(IdentityPerson.id).where( IdentityPerson.id.in_(person_ids), IdentityPerson.employment_status == "active", ) ) ).scalars().all() return set(rows) def _classify_asset( self, asset: AssetItem, *, active_assignees: set[UUID], primary_admin_by_location: dict[UUID, UUID], location_by_office: dict[UUID, UUID], index: LocationIndex, ) -> tuple[str, UUID | None, UUID | None]: """Decide ``(holder_kind, holder_person_id, office_id)``. Priority order: 1. Explicit ``holder_kind='office'`` ⇒ the asset physically sits at that office, so route to the administrator of the location the office is in, and keep ``office_id`` as the display context. 2. ``assigned_to`` points at an active person ⇒ assignee. 3. Otherwise (idle stock, or a holder who has left) resolve the asset's managed ``location`` and route to that location's administrator. 4. Anything else ⇒ IT fallback. The bucket's token is ``location_admin`` (migration 543). It read ``office_admin`` until the recipient stopped being the office's 当地 Admin; a stored enum that names the wrong party costs every later reader a round of "who actually gets this?". Rows written before 543 are still accepted on read — filter on ``HOLDER_KINDS_LOCATION_ADMIN``, never on the bare new value. """ if ( asset.holder_kind == "office" and asset.holder_office_id is not None ): location_id = location_by_office.get(asset.holder_office_id) if location_id is not None: admin_id = primary_admin_by_location.get(location_id) if admin_id is not None: return HOLDER_KIND_LOCATION_ADMIN, admin_id, asset.holder_office_id # The office resolves but its location has no single # accountable administrator — surface to IT with the # office context preserved. return "it_fallback", None, asset.holder_office_id # holder_office_id points at an archived office, or one # whose city names no managed location — fall through to the # asset's own location rather than silently orphaning it. if asset.assigned_to and asset.assigned_to in active_assignees: return "assignee", asset.assigned_to, None location_id = index.resolve_id(asset.location) if location_id is not None: admin_id = primary_admin_by_location.get(location_id) if admin_id is not None: return HOLDER_KIND_LOCATION_ADMIN, admin_id, None return "it_fallback", None, None async def _build_snapshot_items( self, cycle: AssetInventoryCycle, ) -> list[AssetInventoryItem]: scope = cycle.scope_filter or {} assets = await self._query_eligible_assets(scope) if not assets: return [] candidate_assignees = { a.assigned_to for a in assets if a.assigned_to is not None } active_assignees = await self._active_person_ids(candidate_assignees) ( primary_admin_by_location, location_by_office, location_index, ) = await self._resolve_location_admins() items: list[AssetInventoryItem] = [] for asset in assets: kind, holder, office = self._classify_asset( asset, active_assignees=active_assignees, primary_admin_by_location=primary_admin_by_location, location_by_office=location_by_office, index=location_index, ) items.append( AssetInventoryItem( cycle_id=cycle.id, asset_id=asset.id, holder_kind=kind, holder_person_id=holder, office_id=office, asset_tag_snapshot=asset.asset_tag, asset_name_snapshot=asset.asset_name, asset_category_snapshot=asset.category, asset_location_snapshot=asset.location, asset_status_snapshot=asset.status, status="pending", ) ) return items async def preview_cycle( self, cycle_id: UUID, ) -> dict[str, Any]: """Build the snapshot in-memory only; nothing persisted. Lets the admin sanity-check the recipient list before committing. Limited to ``draft`` cycles — replaying a preview on a started cycle would lie about what was actually fanned out (the snapshot is taken once, at start). """ cycle = await self.get_cycle(cycle_id) if cycle.status != "draft": raise ConflictError( "preview is only available for draft cycles", ) return await self._preview_from_filter(cycle.scope_filter or {}) async def preview_filter( self, scope_filter: CycleScopeFilter | None, ) -> dict[str, Any]: """Stateless preview: dry-run the snapshot for a candidate filter without persisting a cycle row. Used by the create-cycle form so the admin sees how many assets (and which holders) match BEFORE clicking submit — critical for "I want to test on just my own laptop" workflows where blindly creating a cycle would fan a Feishu card out to everyone in the company. The output shape matches ``preview_cycle`` so the frontend renders both with the same component. """ # ``mode='json'`` keeps the dict round-trippable through the # JSON column (matches ``create_cycle``) so a downstream caller # that decides to persist this filter doesn't hit # ``TypeError: UUID is not JSON serializable`` from raw UUIDs. filter_dict: dict[str, Any] = ( scope_filter.model_dump(exclude_none=True, mode="json") if scope_filter is not None else {} ) return await self._preview_from_filter(filter_dict) async def _preview_from_filter( self, filter_dict: dict[str, Any], ) -> dict[str, Any]: """Shared engine for cycle-bound and stateless previews. Builds the same ``AssetInventoryItem`` instances ``start_cycle`` would build, but in-memory — nothing is added to the session, so the dry-run is safe to call from any read transaction. """ # _build_snapshot_items only reads cycle.scope_filter; pass a # transient cycle-shaped object rather than building/persisting # one. This keeps the dry-run completely side-effect-free. transient = AssetInventoryCycle( name="__preview__", status="draft", scope_filter=filter_dict, ) items = await self._build_snapshot_items(transient) assignee_items = [i for i in items if i.holder_kind == "assignee"] location_admin_items = [ i for i in items if i.holder_kind in HOLDER_KINDS_LOCATION_ADMIN ] fallback_items = [i for i in items if i.holder_kind == "it_fallback"] recipient_ids = { i.holder_person_id for i in items if i.holder_person_id is not None } # Hydrate sample names so the admin sees who's about to get # pinged. Cap samples to 10 per bucket to keep the response # small even on company-wide previews. sample_ids = list(recipient_ids)[:20] person_rows = ( await self.db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where(IdentityPerson.id.in_(sample_ids)) ) ).all() if sample_ids else [] person_lookup = { p[0]: {"id": str(p[0]), "name": p[1], "email": p[2]} for p in person_rows } sample_assignees: list[dict[str, Any]] = [] seen_assignees: set[UUID] = set() for i in assignee_items: if i.holder_person_id and i.holder_person_id not in seen_assignees: seen_assignees.add(i.holder_person_id) if i.holder_person_id in person_lookup: sample_assignees.append(person_lookup[i.holder_person_id]) if len(sample_assignees) >= 10: break sample_location_admins: list[dict[str, Any]] = [] seen_location_admins: set[UUID] = set() for i in location_admin_items: if i.holder_person_id and i.holder_person_id not in seen_location_admins: seen_location_admins.add(i.holder_person_id) if i.holder_person_id in person_lookup: sample_location_admins.append(person_lookup[i.holder_person_id]) if len(sample_location_admins) >= 10: break sample_fallback = [ { "asset_id": str(i.asset_id), "asset_tag": i.asset_tag_snapshot, "asset_name": i.asset_name_snapshot, "location": i.asset_location_snapshot, } for i in fallback_items[:10] ] return { "total_assets": len(items), "assignee_items": len(assignee_items), "location_admin_items": len(location_admin_items), "it_fallback_items": len(fallback_items), "distinct_recipients": len(recipient_ids), "sample_assignee_persons": sample_assignees, "sample_location_admins": sample_location_admins, "sample_it_fallback_assets": sample_fallback, } async def start_cycle( self, cycle_id: UUID, *, actor_user_id: UUID | None = None, trace_id: str = "", ) -> AssetInventoryCycle: # Lock the cycle row so two admins clicking "Start" at the same # instant can't both run the snapshot — the second waits for the # first to commit and then sees status='active' here and 409s. cycle_row = ( await self.db.execute( select(AssetInventoryCycle) .where(AssetInventoryCycle.id == cycle_id) .with_for_update() ) ).scalar_one_or_none() if cycle_row is None: raise NotFoundError(f"inventory cycle {cycle_id} not found") cycle = cycle_row if cycle.status != "draft": raise ConflictError("only draft cycles can be started") items = await self._build_snapshot_items(cycle) if not items: # Empty snapshot — let the admin know vs silently flipping # to ``active`` with 0 items. Closed empty cycles are # noise; refuse and let the admin adjust scope. raise ConflictError( "no eligible assets matched the cycle scope — " "adjust the scope filter and try again", ) for item in items: self.db.add(item) cycle.status = "active" cycle.started_at = datetime.now(UTC) cycle.total_items = len(items) cycle.pending_count = len(items) cycle.confirmed_count = 0 cycle.issue_count = 0 await self.db.flush() # NOTE: notifications are NOT dispatched here. This method only # performs the (fast, all-DB) state transition while holding the # ``SELECT ... FOR UPDATE`` lock on the cycle row. The launch # fan-out — one Feishu card per holder, hundreds of external HTTP # calls for a company-wide cycle — is done AFTER the caller commits, # off the locked transaction, by # :meth:`dispatch_start_notifications` (run in the background for # the HTTP paths, inline for the twice-a-year cron). Doing it inline # here held the cycle lock + ``identity_persons`` FK locks for # minutes, so retries piled onto the locked row (1205 lock-wait) and # concurrent logins timed out — the 2026-07-14 "开始盘点失败" outage. return cycle async def dispatch_start_notifications( self, cycle_id: UUID, *, actor_user_id: UUID | None = None, trace_id: str = "", ) -> int: """Fan out the launch notification for an already-started cycle. Loads the active cycle + its snapshot items from THIS session (so it is safe to run on a fresh background/worker session, not the request session that created the items) and pushes one card per recipient. Idempotent-ish: a cycle that isn't ``active`` is a no-op, and re-running re-sends (only ever spawned once per successful ``start_cycle``). Returns the recipient count. MUST run on a session that is NOT holding the cycle ``FOR UPDATE`` lock — delivery makes slow external calls and commits per recipient. """ cycle = await self.db.get(AssetInventoryCycle, cycle_id) if cycle is None or cycle.status != "active": return 0 items = ( await self.db.execute( select(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle_id, ) ) ).scalars().all() return await self._dispatch_initial_notifications( cycle, list(items), actor_user_id=actor_user_id, trace_id=trace_id, ) async def cycles_for_launch_reconcile( self, *, grace_minutes: int = _LAUNCH_RECONCILE_GRACE_MINUTES, ) -> list[AssetInventoryCycle]: """Active cycles old enough for the launch-loss reconciler. The grace window keeps the hourly reconciler from racing the just-spawned background drain of a freshly started cycle — inside the window a holder with no outbox row is most likely *about to be* enqueued, not lost. """ cutoff = datetime.now(UTC) - timedelta(minutes=grace_minutes) stmt = select(AssetInventoryCycle).where( AssetInventoryCycle.status == "active", AssetInventoryCycle.started_at.isnot(None), AssetInventoryCycle.started_at <= cutoff, ) return list((await self.db.execute(stmt)).scalars().all()) async def reconcile_launch_notifications( self, cycle_id: UUID, *, actor_user_id: UUID | None = None, trace_id: str = "", ) -> int: """Re-enqueue launch cards that were LOST before phase 1 recorded them. The launch fan-out runs via ``background.spawn`` in the WEB container on the HTTP start paths; a deploy that restarts the container between the start-cycle commit and the end of the enqueue phase drops every holder not yet written to ``notification_outbox``. The ``notification.outbox_retry`` worker only backstops rows that EXIST (phase-2 crashes) — a holder with no row at all was invisible until this reconciler, and re-clicking "Start" 409s on the active-status guard, so nothing else could recover the card (the deploy-kills-drain class of [[reference_long_backfill_deploy_kills_drain]]). Idempotent: a holder counts as "already launched" when ANY ``asset_inventory_notify`` outbox row exists for them on this cycle, regardless of delivery status. Holders whose items are all confirmed/issue-reported no longer need the launch card and are skipped, so the missing set converges to zero. Returns the number of holders (re-)enqueued. MUST run off any transaction holding the cycle ``FOR UPDATE`` lock — the fan-out it delegates to commits per recipient. """ cycle = await self.db.get(AssetInventoryCycle, cycle_id) if cycle is None or cycle.status != "active": return 0 items = ( await self.db.execute( select(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle_id, AssetInventoryItem.status == "pending", AssetInventoryItem.holder_person_id.isnot(None), ) ) ).scalars().all() if not items: return 0 # Holders that already have a launch row for this cycle — # compare on the launch template only, so reminder/escalation # rows (different template_key) never mask a lost launch card. # ``related_resource_id`` is CHAR(36); probe with the string # form like the other outbox idempotency checks do. already_ids = { row for row in ( await self.db.execute( select(NotificationOutbox.recipient_person_id).where( NotificationOutbox.template_key == _NOTIFICATION_TEMPLATE_KEY, NotificationOutbox.related_resource_id == str(cycle_id), NotificationOutbox.recipient_person_id.isnot(None), ).distinct() ) ).scalars().all() } missing_items = [ i for i in items if i.holder_person_id not in already_ids ] if not missing_items: return 0 logger.warning( "inventory launch reconcile: cycle %s has %d holder(s) with " "pending items and NO launch outbox row — re-enqueueing " "(lost enqueue, e.g. deploy killed the background drain).", cycle_id, len({i.holder_person_id for i in missing_items}), ) return await self._dispatch_initial_notifications( cycle, missing_items, actor_user_id=actor_user_id, trace_id=trace_id, ) # ── Notifications ───────────────────────────────────────────────── async def _dispatch_initial_notifications( self, cycle: AssetInventoryCycle, items: list[AssetInventoryItem], *, actor_user_id: UUID | None, trace_id: str, ) -> int: """One Feishu card per recipient (across both buckets). Durable two-phase, per-recipient-committed fan-out — must run on a session that is NOT holding the cycle ``FOR UPDATE`` lock: * **Phase 1 — enqueue.** Write every recipient's ``notification_outbox`` row PENDING and commit each individually. After this phase every intended card is durably recorded, so a crash during phase 2 (deploy restart mid-drain — the background task lives in the web process) leaves the un-sent rows for the hourly ``notification.outbox_retry`` worker: guaranteed delivery holds. Each row is committed with ``next_attempt_at`` a fence interval in the future (see :data:`_OUTBOX_RETRY_FENCE`) so an hourly retry tick landing inside a multi-minute phase-2 drain cannot claim the same PENDING rows and double-send them; phase 2 drains its own explicit row list, unaffected by the fence. * **Phase 2 — deliver.** Attempt each pending row's channels and commit per recipient. A per-row failure just leaves that row PENDING for the retry worker. Per-recipient commit (never one big transaction) keeps the shared ``identity_persons`` FK locks each outbox INSERT takes to a sub-millisecond window, so the fan-out can never brown-out logins the way the old single-transaction dispatch did (see [[reference_outbox_fk_lock_login_outage]]). Returns the recipient count. """ # Group by recipient. it_fallback items have no recipient and # are silently parked in the admin bucket — no notification. by_person: dict[UUID, dict[str, list[AssetInventoryItem]]] = ( defaultdict(lambda: {"assignee": [], "location_admin": []}) ) for item in items: if item.holder_person_id is None: continue by_person[item.holder_person_id][_bucket_key(item.holder_kind)].append(item) if not by_person: return 0 person_rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(by_person.keys())), ) ) ).scalars().all() person_lookup = {p.id: p for p in person_rows} portal_url = await self._portal_url() notif = NotificationService(self.db) # Phase 1: enqueue every recipient's row (durable), commit per row. enqueued: list[tuple[Any, NotificationEnvelope]] = [] for person_id, buckets in by_person.items(): person = person_lookup.get(person_id) if person is None or not person.work_email: continue env = self._build_envelope( person, cycle, len(buckets["assignee"]), len(buckets["location_admin"]), template_key=_NOTIFICATION_TEMPLATE_KEY, portal_url=portal_url, ) try: outbox = await notif.enqueue(env) # Fence the row out of the hourly retry picker while # phase 2 below is still draining — otherwise a retry # tick landing mid-drain claims the same PENDING rows # (its picker sorts next_attempt_at=NULL FIRST) and # sends duplicates. If this process dies before the # row is delivered, the worker picks it up after the # fence expires. See _OUTBOX_RETRY_FENCE. outbox.next_attempt_at = ( datetime.now(UTC) + _OUTBOX_RETRY_FENCE ) await self.db.commit() except Exception: # noqa: BLE001 # begin_nested already rolled back the INSERT + expunged # the row; roll back the outer txn so the session stays # usable for the next recipient. A missed enqueue here is # rare (transient DB error) and only costs one card. await self.db.rollback() logger.exception( "failed to enqueue inventory notification for %s", person.id, ) continue enqueued.append((outbox, env)) # Phase 2: deliver each pending row, commit per row. A failure # leaves the row PENDING → hourly ``notification.outbox_retry``. for outbox, env in enqueued: try: await notif.deliver( outbox, env, trace_id=trace_id, actor_user_id=actor_user_id, ) except Exception: # noqa: BLE001 logger.exception( "failed to deliver inventory notification to %s", env.recipient.id, ) await self.db.commit() return len(enqueued) async def _portal_url(self) -> str: # Platform setting ``app.frontend_base_url`` is the public-facing # frontend origin (``https://helix.example.com``). Email and # Feishu cards prepend it so recipients get a clickable link # instead of a bare path. Reads the SAME key # ``NotificationService._resolve_absolute_url`` reads — the # parallel ``portal.base_url`` it used to consult drifted to # ``http://localhost:3000`` on prod and leaked into every # inventory reminder body / Feishu card, even though the action # button rendered correctly. from app.integrations.service import get_platform_setting v = await get_platform_setting( "app.frontend_base_url", default="", session=self.db, ) return str(v or "").rstrip("/") def _build_envelope( self, person: IdentityPerson, cycle: AssetInventoryCycle, personal_n: int, office_n: int, *, template_key: str, portal_url: str = "", ) -> NotificationEnvelope: url = f"{portal_url}/me/inventory/{cycle.id}" title = f"[资产盘点] {cycle.name}" # Greeting fallback: HRIS rows occasionally land with NULL # ``display_name`` (legacy seeds, mid-sync gaps). Bare "你好 ," # in a Feishu card reads as a bug — back off to the # work_email local-part, which we know exists because # `_dispatch_initial_notifications` / `send_cycle_reminder` # already skip recipients with no work_email. greeting_name = ( person.display_name or (person.work_email.split("@", 1)[0] if person.work_email else "") ) lines = [ f"你好 {greeting_name},", "", f"资产盘点周期『{cycle.name}』需要你确认。", ] if personal_n: lines.append(f" · 个人持有资产: {personal_n} 件") if office_n: lines.append(f" · 办公室待盘资产: {office_n} 件") if cycle.due_at: lines.append( f" · 截止日期: {cycle.due_at.strftime('%Y-%m-%d')}", ) lines += [ "", "进入 Helix 完成确认或反馈问题。", ] body_text = "\n".join(lines) return NotificationEnvelope( recipient=person, title=title, body_text=body_text, channels=["feishu", "email"], template_key=template_key, context={ "person_name": greeting_name, "cycle_name": cycle.name, "personal_count": personal_n, "office_count": office_n, "due_date": ( cycle.due_at.strftime("%Y-%m-%d") if cycle.due_at else "" ), "portal_url": url, }, url=url, related_resource_type="asset_inventory_cycle", related_resource_id=cycle.id, ) async def send_cycle_reminder( self, cycle_id: UUID, *, actor_user_id: UUID | None = None, trace_id: str = "", force: bool = False, working_hours_only: bool = False, notify_manager: bool = False, notify_hrbp: bool = False, ) -> ReminderOutcome: """Push a reminder to recipients with pending items. Per-recipient daily cadence: each recipient receives at most one card per *local* day, dedup'd via ``AssetInventoryItem.reminded_at`` (UTC) compared against the local-midnight UTC instant for that recipient's timezone. ``working_hours_only=True`` (set by the scheduled cron) only delivers to recipients whose current local time falls within Mon–Fri 10:00–17:59. ``force=True`` bypasses both the working-hours gate and the per-day dedup — used by the admin "立即催办" button. Anti-double-send: the cycle row-lock is held ONLY for a fast claim phase — every target recipient's ``reminded_at`` is stamped and committed ONCE up front, and the slow external sends run afterwards against the claimed set. A concurrent caller (cron tick racing an admin 立即催办) acquires the lock next, sees the stamps, and claims zero rows. ``force=True`` goes through the same claim: it ignores the per-day dedup but skips recipients stamped within :data:`_FORCE_RECLAIM_GRACE`, so two overlapping force calls cannot both send. Delivery itself is two-phase per recipient (outbox row committed BEFORE the external send) so no ``identity_persons`` FK lock is ever held across a Feishu/email call. ``notify_manager`` / ``notify_hrbp`` (admin opt-in on the manual remind, default off) additionally send ONE aggregated nudge digest per Lead / HRBP that lists their still-unresponsive people — so the cron (which never sets these) stays employee-only, and the manager/HRBP escalation never fans out one-card-per-employee. Returns a :class:`ReminderOutcome` with the distinct-recipient counts for each leg. """ # ── Phase A: claim under the cycle lock ────────────────────── # Row-lock serialises the scheduled reminder handler and an # admin clicking the "Remind" button. The per-item # ``reminded_at`` stamp (written for ALL targets before any # send, committed once) is the claim marker a concurrent # caller observes. cycle_row = ( await self.db.execute( select(AssetInventoryCycle) .where(AssetInventoryCycle.id == cycle_id) .with_for_update() # populate_existing: the reminder handler first loads these # cycles via cycles_due_for_reminder() into this same session, # so without it this locked re-read returns the stale # identity-map copy and the status guard below could act on # pre-lock state (the locked-get identity-map trap #1249 swept). .execution_options(populate_existing=True) ) ).scalar_one_or_none() if cycle_row is None: raise NotFoundError(f"inventory cycle {cycle_id} not found") cycle = cycle_row if cycle.status != "active": raise ConflictError( "only active cycles can have reminders dispatched", ) # Pull all pending items with a recipient. rows = ( await self.db.execute( select(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle.id, AssetInventoryItem.status == "pending", AssetInventoryItem.holder_person_id.isnot(None), ) ) ).scalars().all() now = datetime.now(UTC) if not rows: if cycle.reminder_sent_at is None: cycle.reminder_sent_at = now await self.db.commit() return ReminderOutcome() by_person: dict[UUID, dict[str, list[AssetInventoryItem]]] = ( defaultdict(lambda: {"assignee": [], "location_admin": []}) ) for item in rows: assert item.holder_person_id is not None by_person[item.holder_person_id][_bucket_key(item.holder_kind)].append(item) person_rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(by_person.keys())), ) ) ).scalars().all() person_lookup = {p.id: p for p in person_rows} def _naive_utc(dt: datetime) -> datetime: return dt.replace(tzinfo=None) if dt.tzinfo else dt # Third tuple element = the per-item pre-claim ``reminded_at`` # values, so Phase B can COMPENSATE (restore the stamp) if the # enqueue fails — otherwise the claim commit below would leave the # recipient marked reminded-today with no outbox row and the # per-day dedup would silently suppress them until tomorrow. claimed: list[ tuple[ IdentityPerson, dict[str, list[AssetInventoryItem]], list[tuple[AssetInventoryItem, datetime | None]], ] ] = [] for person_id, buckets in by_person.items(): person = person_lookup.get(person_id) if person is None or not person.work_email: continue # Pre-launch P1 fix (Round 45, CR-409): skip leavers / # leaver_pending recipients. The borrow-overdue cron was # fixed (Round-44) but the inventory reminder cron was # missed — leaver employees in their last few days kept # being nagged about inventory items, on top of all the # offboarding pushes. person_status = (person.employment_status or "").lower() if person_status in {"leaver_pending", "left"}: continue if force: # force ignores the per-day dedup, but a stamp younger # than the grace window means another caller claimed # this recipient moments ago and may still be mid-send # — re-claiming would double-send (two overlapping # 立即催办 clicks, or 立即催办 racing the cron). claim_cutoff = _naive_utc(now) - _FORCE_RECLAIM_GRACE recently_claimed = any( item.reminded_at is not None and _naive_utc(item.reminded_at) >= claim_cutoff for items in buckets.values() for item in items ) if recently_claimed: continue else: tz = resolve_timezone(person) if working_hours_only and not is_local_working_time(now, tz): continue # Per-local-day dedup. Any pending item for this person # already reminded since their local midnight → skip. # MySQL/TiDB ``DATETIME`` columns drop tzinfo, so # ``item.reminded_at`` comes back naive on a fresh # load even though we wrote it from ``datetime.now(UTC)``; # the same row read through SQLAlchemy's identity map # within the writing transaction stays aware. PR #440 # compared the two against an aware right-hand side # and raised ``TypeError: can't compare offset-naive # and offset-aware`` on every fresh worker tick. The # local pytest fixture missed it because the identity # map kept the aware copy in-memory. Normalize both # sides to naive UTC — every value here is UTC by # construction (``datetime.now(UTC)``), so dropping # tzinfo loses no information. today_local_start_utc_naive = local_start_of_day( now, tz, ).replace(tzinfo=None) already_today = any( item.reminded_at is not None and _naive_utc(item.reminded_at) >= today_local_start_utc_naive for items in buckets.values() for item in items ) if already_today: continue # ── Cadence taper (2026-08-30) ────────────────────── # The per-local-day dedup above caps this at one card a # day; it does not cap how many days in a row. Thirty # days of a cycle meant up to thirty cards, and # production sat at 16.7 per person per month — the # single loudest thing this platform pushed at # employees. The ladder keeps the first few daily (most # people answer then, and slowing those would cost real # response time for noise nobody felt yet) and stretches # only the long tail, which is where all the volume is. # Nobody is ever dropped: an unanswered cycle keeps # producing reminders, just weekly instead of daily. # ``force`` (admin 立即催办) deliberately bypasses this — # a human choosing to chase someone right now is not # noise, and the FORCE_RECLAIM_GRACE above already stops # a double-send. # ``taper_holds`` rather than ``should_send``: the # per-local-day dedup above already governs the daily # rungs, and stacking a second wall-clock day test on it # would drop a card whenever yesterday's tick ran late # and today's ran early — a noise change quietly # becoming a delivery change. if await cadence.taper_holds( self.db, recipient_id=person_id, template_key=_REMINDER_TEMPLATE_KEY, subject_key=str(cycle_id), now=now, ): continue # Claim: stamp every pending item BEFORE any send happens, # capturing the prior value so a failed enqueue can restore it. prior_stamps: list[tuple[AssetInventoryItem, datetime | None]] = [] for item_list in buckets.values(): for item in item_list: prior_stamps.append((item, item.reminded_at)) item.reminded_at = now claimed.append((person, buckets, prior_stamps)) # ``reminder_sent_at`` is "first reminder timestamp" (audit / # UI display only); the per-day dedup lives on ``reminded_at``. # Set it with the first claim and leave it alone afterwards. if claimed and cycle.reminder_sent_at is None: cycle.reminder_sent_at = now # Single claim commit: releases the cycle FOR UPDATE lock and # publishes every stamp to concurrent callers BEFORE the slow # external fan-out below. No lock is ever held across a Feishu # send (the multi-minute-lock login brown-out of 2026-07-14, # see [[reference_outbox_fk_lock_login_outage]]), yet the claim # itself stays atomic — a second caller entering mid-fan-out # finds the whole target set already stamped, not just the # recipients sent so far. await self.db.commit() # ── Phase B: deliver to the claimed set ────────────────────── portal_url = await self._portal_url() notif = NotificationService(self.db) sent = 0 for person, buckets, prior_stamps in claimed: env = self._build_envelope( person, cycle, len(buckets["assignee"]), len(buckets["location_admin"]), template_key=_REMINDER_TEMPLATE_KEY, portal_url=portal_url, ) # Two-phase per recipient: durably record + COMMIT the # outbox row first, then attempt the external send, then # commit the delivery bookkeeping. The outbox INSERT takes # a share-lock on the recipient's ``identity_persons`` row # (FK) — committing before the send keeps that lock # sub-millisecond instead of holding it for the duration of # a degraded Feishu call (single-row-scope login-1205 risk). # A crash after the enqueue-commit leaves the row PENDING # for the hourly retry worker once the fence expires. try: outbox = await notif.enqueue(env) outbox.next_attempt_at = ( datetime.now(UTC) + _OUTBOX_RETRY_FENCE ) # Advance the ladder in the SAME transaction as the # outbox row. Recording it after the external send # would repeat the 2026-08-13 receipt bug in a new # place: the card goes out, delivery wobbles, the # count never lands, and tomorrow's tick sends again # because as far as the ladder knows nothing ever # happened. The receipt means "we decided to ask and # durably queued it" — exactly what the outbox row # guarantees, and exactly what the compensation path # below rolls back when the enqueue fails. await cadence.record_sent( self.db, recipient_id=person.id, template_key=_REMINDER_TEMPLATE_KEY, subject_key=str(cycle_id), now=now, ) await self.db.commit() except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "failed to enqueue inventory reminder for %s", person.id, ) # Compensate: the claim stamp was durably committed in # Phase A, but with no outbox row the per-day dedup would # suppress this recipient until tomorrow. Restore the # prior ``reminded_at`` so a later run today re-attempts. # A concurrent caller that already skipped this recipient # won't re-send (it passed), so this stays exactly-once. # (``reminder_sent_at`` is a display-only "first reminder" # marker — left as-is; only the per-recipient stamp drives # dedup and retry.) try: for item, prior in prior_stamps: item.reminded_at = prior await self.db.commit() except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "failed to restore reminder claim for %s", person.id, ) continue try: await notif.deliver( outbox, env, trace_id=trace_id, actor_user_id=actor_user_id, ) except Exception: # noqa: BLE001 logger.exception( "failed to deliver inventory reminder to %s", person.id, ) await self.db.commit() sent += 1 outcome = ReminderOutcome(recipients_notified=sent) # Optional Lead / HRBP escalation. The pool is every *valid* # pending holder (non-leaver, has work_email) — NOT gated by the # per-day dedup / working-hours window that throttles the # employee's own card, because this is a separate admin-initiated # "please chase them" push, not a repeat nag at the employee. if notify_manager or notify_hrbp: valid_pending = [ p for p in person_lookup.values() if p.work_email and (p.employment_status or "").lower() not in {"leaver_pending", "left"} ] ( outcome.managers_notified, outcome.hrbps_notified, ) = await self._send_reminder_escalations( cycle, valid_pending, portal_url=portal_url, notify_manager=notify_manager, notify_hrbp=notify_hrbp, trace_id=trace_id, actor_user_id=actor_user_id, ) # Belt-and-braces: everything above is already committed (claim # commit + per-recipient / per-digest commits); this only picks # up incidental session state (e.g. HRBP-resolution cache writes). await self.db.commit() return outcome async def _send_reminder_escalations( self, cycle: AssetInventoryCycle, pending_persons: list[IdentityPerson], *, portal_url: str, notify_manager: bool, notify_hrbp: bool, trace_id: str, actor_user_id: UUID | None, ) -> tuple[int, int]: """Send aggregated nudge digests to Leads and/or HRBPs. One card per distinct Lead / HRBP listing the unresponsive people related to them — never one per employee. Returns ``(managers_notified, hrbps_notified)``. Best-effort: a single dispatch / resolve failure is logged and skipped, never aborts the others (mirrors the employee-reminder loop above). """ if not pending_persons: return (0, 0) notif = NotificationService(self.db) async def _send_digest( recipient: IdentityPerson, people: list[IdentityPerson], ) -> bool: # Never nudge someone about themselves, and skip a # Lead/HRBP who is themselves a leaver / has no inbox. targets = [p for p in people if p.id != recipient.id] if not targets: return False if not recipient.work_email: return False if (recipient.employment_status or "").lower() in { "leaver_pending", "left", }: return False env = self._build_escalation_envelope( recipient, cycle, targets, portal_url=portal_url, ) # Two-phase per digest — same lock hygiene as the employee # loop: commit the outbox row (and its ``identity_persons`` # FK share-lock) BEFORE the external Feishu/email send, so a # degraded send never holds the recipient's person row. try: outbox = await notif.enqueue(env) outbox.next_attempt_at = ( datetime.now(UTC) + _OUTBOX_RETRY_FENCE ) await self.db.commit() except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "failed to enqueue inventory escalation for %s", recipient.id, ) return False try: await notif.deliver( outbox, env, trace_id=trace_id, actor_user_id=actor_user_id, ) except Exception: # noqa: BLE001 logger.exception( "failed to deliver inventory escalation to %s", recipient.id, ) await self.db.commit() return True managers_notified = 0 hrbps_notified = 0 if notify_manager: by_manager: dict[UUID, list[IdentityPerson]] = defaultdict(list) for p in pending_persons: if p.manager_person_id: by_manager[p.manager_person_id].append(p) mgr_rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(by_manager.keys())), ) ) ).scalars().all() if by_manager else [] mgr_lookup = {m.id: m for m in mgr_rows} for mgr_id, people in by_manager.items(): mgr = mgr_lookup.get(mgr_id) if mgr is None: continue if await _send_digest(mgr, people): managers_notified += 1 if notify_hrbp: # Resolve each pending person's HRBP. ``resolve_hrbp_for_person`` # short-circuits on the SF-authoritative / cached column (one # PK lookup) and only walks Feishu for the unresolved tail; # it never raises. Group by the resolved HRBP so each HRBP # gets a single digest. from app.integrations.feishu_hrbp import resolve_hrbp_for_person by_hrbp: dict[UUID, list[IdentityPerson]] = defaultdict(list) for p in pending_persons: resolution = await resolve_hrbp_for_person(self.db, p) if resolution.person_id is not None: by_hrbp[resolution.person_id].append(p) hrbp_rows = ( await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(by_hrbp.keys())), ) ) ).scalars().all() if by_hrbp else [] hrbp_lookup = {h.id: h for h in hrbp_rows} for hrbp_id, people in by_hrbp.items(): hrbp = hrbp_lookup.get(hrbp_id) if hrbp is None: continue if await _send_digest(hrbp, people): hrbps_notified += 1 return (managers_notified, hrbps_notified) def _build_escalation_envelope( self, recipient: IdentityPerson, cycle: AssetInventoryCycle, pending_people: list[IdentityPerson], *, portal_url: str, ) -> NotificationEnvelope: url = f"{portal_url}/dashboard" title = f"[资产盘点·协助提醒] {cycle.name}" greeting_name = ( recipient.display_name or ( recipient.work_email.split("@", 1)[0] if recipient.work_email else "" ) ) def _display(p: IdentityPerson) -> str: base = ( p.display_name or (p.work_email.split("@", 1)[0] if p.work_email else str(p.id)) ) return f"{base}({p.employee_no})" if p.employee_no else base names = [_display(p) for p in pending_people] # Pre-format the bullet list once; both the template render path # (Feishu / email pull the template body and inject {pending_names}) # and the envelope fallback body use the same string. pending_names = "\n".join(f" · {n}" for n in names) lines = [ f"你好 {greeting_name},", "", f"资产盘点周期『{cycle.name}』进行中,以下与你相关的 " f"{len(names)} 位同事仍未反馈其资产情况:", "", pending_names, "", "烦请协助提醒他们尽快在 Helix 完成确认。", ] body_text = "\n".join(lines) return NotificationEnvelope( recipient=recipient, title=title, body_text=body_text, channels=["feishu", "email"], template_key=_ESCALATION_TEMPLATE_KEY, context={ "person_name": greeting_name, "cycle_name": cycle.name, "pending_count": len(names), "pending_names": pending_names, "portal_url": url, }, url=url, related_resource_type="asset_inventory_cycle", related_resource_id=cycle.id, ) # ── Item operations ─────────────────────────────────────────────── async def list_items( self, cycle_id: UUID, *, holder_kind: str | None = None, status: str | None = None, holder_person_id: UUID | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[AssetInventoryItem], int]: # Confirm cycle exists for a clean 404 vs a silent empty list. await self.get_cycle(cycle_id) stmt = select(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle_id, ) count_stmt = select(func.count()).select_from(AssetInventoryItem).where( AssetInventoryItem.cycle_id == cycle_id, ) if holder_kind: if holder_kind not in ALLOWED_HOLDER_KINDS: raise ValidationError(f"unknown holder_kind: {holder_kind}") stmt = stmt.where(AssetInventoryItem.holder_kind == holder_kind) count_stmt = count_stmt.where( AssetInventoryItem.holder_kind == holder_kind, ) if status: if status not in ALLOWED_ITEM_STATUSES: raise ValidationError(f"unknown status: {status}") stmt = stmt.where(AssetInventoryItem.status == status) count_stmt = count_stmt.where( AssetInventoryItem.status == status, ) if holder_person_id is not None: stmt = stmt.where( AssetInventoryItem.holder_person_id == holder_person_id, ) count_stmt = count_stmt.where( AssetInventoryItem.holder_person_id == holder_person_id, ) stmt = stmt.order_by(asc(AssetInventoryItem.holder_kind), asc(AssetInventoryItem.created_at)) total = (await self.db.execute(count_stmt)).scalar() or 0 offset = max(0, (page - 1) * page_size) stmt = stmt.offset(offset).limit(page_size) rows = (await self.db.execute(stmt)).scalars().all() return list(rows), total async def list_pending_for_person( self, person_id: UUID, *, cycle_id: UUID | None = None, ) -> list[AssetInventoryItem]: """All pending items routed to ``person_id`` across active cycles (or a specific cycle when ``cycle_id`` is given). Used by the employee-facing /me/inventory page. """ stmt = select(AssetInventoryItem).where( AssetInventoryItem.holder_person_id == person_id, ) if cycle_id is not None: stmt = stmt.where(AssetInventoryItem.cycle_id == cycle_id) else: # Restrict to currently-active cycles when none specified. stmt = stmt.join( AssetInventoryCycle, AssetInventoryCycle.id == AssetInventoryItem.cycle_id, ).where(AssetInventoryCycle.status == "active") stmt = stmt.order_by(asc(AssetInventoryItem.holder_kind)) rows = (await self.db.execute(stmt)).scalars().all() return list(rows) async def get_item( self, item_id: UUID, *, lock: bool = False, ) -> AssetInventoryItem: # ``lock=True`` adds SELECT ... FOR UPDATE so respond_to_item # serialises concurrent responders on the same item — without # this two writers (e.g. an admin and the holder racing in the # same second) both observe status='pending' and silently # overwrite each other's response_type / responded_by, breaking # the "first responder wins" invariant. if lock: row = ( await self.db.execute( select(AssetInventoryItem) .where(AssetInventoryItem.id == item_id) .with_for_update() ) ).scalar_one_or_none() else: row = await self.db.get(AssetInventoryItem, item_id) if row is None: raise NotFoundError(f"inventory item {item_id} not found") return row async def respond_to_item( self, item_id: UUID, *, actor_person_id: UUID, is_admin: bool, response_type: str, response_notes: str | None = None, new_location: str | None = None, evidence_photos: list[str] | None = None, ) -> AssetInventoryItem: # Lock the item row first so two concurrent responders on the # same item serialise (the cycle lock below alone only protects # counter consistency — not item.response_type / responded_by). item = await self.get_item(item_id, lock=True) # Re-respond ("修改回应") is exposed by /me/inventory while the # cycle is still active and the admin hasn't yet acted on the # prior response. Bounded so we never have to undo side-effects # we can't undo cleanly: # * admin-resolved issues are frozen (admin already moved on). # * a prior 'transferred' response mutated the canonical # asset row + emitted AssetOperation/History; reverting # belongs to the asset transfer flow, not /respond. if item.status == "pending": pass elif item.status in ("confirmed", "issue_reported"): if item.issue_status == "resolved": raise ConflictError( f"item {item_id} has been resolved by IT; " "ask an admin to re-open before editing the response", ) if item.response_type == "transferred": raise ConflictError( f"item {item_id} was previously recorded as " "'transferred' — the asset row has already moved; " "ask an admin to undo via the asset transfer flow " "before editing.", ) else: raise ConflictError( f"item {item_id} has status {item.status!r} and " "cannot be modified", ) # Serialise concurrent responders on the same cycle: take a row # lock on the cycle BEFORE recomputing counters, so two # employees responding at the same instant don't end up with a # split-brain count. ``with_for_update`` translates to MySQL's # ``SELECT ... FOR UPDATE`` and holds until the outer commit. cycle_row = ( await self.db.execute( select(AssetInventoryCycle) .where(AssetInventoryCycle.id == item.cycle_id) .with_for_update() ) ).scalar_one_or_none() if cycle_row is None: raise NotFoundError(f"inventory cycle {item.cycle_id} not found") cycle = cycle_row if cycle.status != "active": raise ConflictError( "responses can only be recorded against active cycles", ) # Authorisation: holder, or any admin (admins respond on behalf # of the it_fallback bucket and may also override on behalf of # an absent recipient). if not is_admin and item.holder_person_id != actor_person_id: raise ForbiddenError("Not authorised to respond to this item.") # "IT 帮忙确认" — an admin answering for the routed holder. The # row will read as confirmed even though the person it was sent # to never touched it, so the *basis* for that confirmation must # be on the record: which channel the holder replied through, # who physically sighted the device, etc. Required only for the # proxy case — an admin answering the ``it_fallback`` bucket # (no holder) or their own item is the normal path and keeps # notes optional. is_proxy_response = ( is_admin and item.holder_person_id is not None and item.holder_person_id != actor_person_id ) if is_proxy_response and not (response_notes or "").strip(): raise ValidationError( "responding on behalf of the routed holder requires " "response_notes — record how the asset was verified", ) allowed = RESPONSE_TYPES_BY_HOLDER_KIND.get(item.holder_kind, ()) if response_type not in allowed: raise ValidationError( f"response_type {response_type!r} not allowed for " f"holder_kind={item.holder_kind!r}", ) if response_type == "moved_to" and not new_location: raise ValidationError( "moved_to response requires new_location", ) # Defence-in-depth: the route layer's Pydantic schema enforces # these too, but the service is reachable directly from tests # / scripts / future internal callers — re-validate the # photo-key shape here so bad inputs can't leak past the # schema boundary and land orphan keys / oversize JSON on # ``asset_inventory_items.evidence_photos``. Shared helper so # the route and service stay in lock-step. try: evidence_photos = sanitize_evidence_photo_keys(evidence_photos) except ValueError as exc: raise ValidationError(str(exc)) from exc if response_type == "not_mine" and not evidence_photos: raise ValidationError( "not_mine response requires at least one evidence photo", ) if response_type == "not_mine" and evidence_photos: from app.core.storage_uploads import mark_uploads_referenced await mark_uploads_referenced( self.db, object_keys=evidence_photos, person_id=actor_person_id, allowed_purposes=frozenset({"inventory_evidence"}), resource_type="asset_inventory_item", resource_id=str(item.id), ) is_confirming = response_type in CONFIRMING_RESPONSES new_status = "confirmed" if is_confirming else "issue_reported" item.status = new_status item.response_type = response_type item.response_notes = response_notes or None # Auxiliary pointers are only meaningful for the response_type # that requires them. ``transferred`` is no longer an accepted # response (user-to-user moves are asset-admin-only via the # dedicated transfer endpoints), so a new response never carries # a transfer target — always clear it. ``new_location`` is kept # only for ``moved_to``. item.transferred_to_id = None item.new_location = ( (new_location or None) if response_type == "moved_to" else None ) # Photos are only meaningful on the not_mine branch — drop any # caller-supplied list on the other branches so a stray payload # doesn't leave orphan keys hanging off a confirmed-mine row. item.evidence_photos = ( list(evidence_photos) if (response_type == "not_mine" and evidence_photos) else None ) item.responded_at = datetime.now(UTC) item.responded_by = actor_person_id if not is_confirming: # Default issue queue position. ``open`` until an admin # marks it ``resolved`` via resolve_issue. item.issue_status = "open" else: item.issue_status = None item.issue_resolved_at = None item.issue_resolved_by = None item.issue_resolution_notes = None # Denormalise onto the asset row so the asset list page can # render "最近盘点" without joining. # Pre-launch P1 fix (Round 45, CR-415): take row-level lock so # a concurrent admin change_assignee doesn't race with the # inventory transfer write. Without the lock, two operations # rows can land pointing at different recipients on the same # second; the asset.assigned_to becomes whichever transaction # commits last with no consistent ordering. asset = ( await self.db.execute( select(AssetItem) .where(AssetItem.id == item.asset_id) .with_for_update() ) ).scalar_one_or_none() if asset is not None and asset.deleted_at is None: asset.last_inventory_at = item.responded_at asset.last_inventory_cycle_id = cycle.id asset.last_inventory_status = new_status # NOTE: inventory responses never move ownership. A holder who # no longer has an asset reports ``not_mine``; the actual # user-to-user transfer is an asset-admin-only operation done # through ``POST /assets/{id}/transfer`` / ``change-assignee`` # (require_asset_admin). See RESPONSE_TYPES_BY_HOLDER_KIND. # ``moved_to`` response — the asset stays with the same # holder but has physically relocated (e.g. Beijing → # Shanghai). Without this write-back ``asset.location`` # would stay at the old city, the asset list page would # show the stale value, AND the next cycle's # ``_classify_asset`` city-based routing would re-route to # the original location_admin, which would re-report # "moved" — an infinite drift loop. Mirror the # ``transferred`` audit pair (no AssetOperation, since # there's no user-to-user motion; AssetHistory carries the # location change for the asset timeline). if ( response_type == "moved_to" and new_location and new_location != asset.location ): # ``new_location`` arrived from the holder's # inventory-response form (free-text). Route through # the managed resolver so an inventory entry like # ``Santa Clara`` lands as ``美国`` rather than # reintroducing a city-level string the dropdown # can't filter. If unresolvable, skip the write # entirely — the cycle will re-prompt the holder. _inv_resolver = await ManagedLocationResolver.load(self.db) _inv_resolved = _inv_resolver.resolve(new_location) if _inv_resolved is not None and _inv_resolved != asset.location: old_location = asset.location now = datetime.now(UTC) asset.location = _inv_resolved asset.updated_at = now # The raw holder-submitted value (HRIS-derived # ``work_location_name`` like ``Santa Clara`` / # ``北京市`` / ``Würzburg``) is INTENTIONALLY # absent from the history notes — Round-5 review # high: ``AssetHistory.notes`` is plain text # surfaced to non-admin readers via # ``GET /assets/{id}`` (in-scope) and the # management-tier ``GET /assets/history`` # endpoint, so painting the raw HRIS string there # leaks the same data the operation.detail # ``admin_view`` gate already protects. Mirror # the ``update_asset`` rejection pattern: keep # the canonical transition in notes (admins + # holders see what happened), surface the raw # value via ``logger.warning`` for admin # correlation in structured logs only. if _inv_resolved != new_location: logger.warning( "inventory_service.respond_to_item: " "normalised holder-supplied location %r " "to managed value %r on asset_id=%s " "(cycle_id=%s)", new_location, _inv_resolved, asset.id, cycle.id, ) self.db.add(AssetHistory( asset_id=asset.id, action="location_changed", operator_id=actor_person_id, notes=( f"Inventory cycle {cycle.id}: " f"{old_location or '∅'} → {_inv_resolved}" ), )) # ``damaged`` response — surface the hardware health # state on the asset row so the asset list page renders # the fault badge and ops cleanup paths (offboarding # scanner, recycle-to-office) treat the row as needing # repair attention. The inventory flow is the only # employee-facing surface that surfaces hardware faults # at scale; without this denormalisation the fault stays # buried in inventory_items.response_type. We write # ``"faulty"`` — the canonical value in # ``ALLOWED_DEVICE_STATUSES`` (normal / faulty / # under_repair / pending_repair). A separate value # ``"damaged"`` would bypass the asset-edit validator and # render as an unknown badge on the asset list. Don't # unset ``device_status`` when the holder flips back to # ``in_use`` — clearing repair history requires the # explicit "mark resolved / completed repair" action. if ( response_type == "damaged" and asset.device_status not in ("faulty", "under_repair") ): asset.device_status = "faulty" asset.updated_at = datetime.now(UTC) self.db.add(AssetHistory( asset_id=asset.id, action="reported_faulty", operator_id=actor_person_id, notes=( f"Inventory cycle {cycle.id}: holder reported " f"damaged condition." ), )) await self.db.flush() # Recompute counters atomically from the items table to avoid # a Python-level read-modify-write race when two recipients # respond concurrently in different transactions. The # ``ix_asset_inv_items_cycle_status`` index makes the GROUP BY # fast (≤ 4 rows scanned per cycle). await self._recompute_counters(cycle.id) await self.db.refresh(cycle) return item async def resolve_issue( self, item_id: UUID, *, actor_person_id: UUID, notes: str | None = None, ) -> AssetInventoryItem: # Row-lock the item so two admins clicking Resolve in the same # instant serialise — without the lock both threads see # ``issue_status='open'``, both write ``resolved`` + their own # actor_id / notes, and the loser's notes are silently lost # (same shape as the respond_to_item race we already fixed). item = await self.get_item(item_id, lock=True) if item.status != "issue_reported": raise ConflictError( "only issue_reported items can be resolved", ) if item.issue_status == "resolved": return item item.issue_status = "resolved" item.issue_resolved_at = datetime.now(UTC) item.issue_resolved_by = actor_person_id item.issue_resolution_notes = notes or None await self.db.flush() return item async def close_cycle( self, cycle_id: UUID, *, actor_person_id: UUID | None = None, ) -> AssetInventoryCycle: cycle_row = ( await self.db.execute( select(AssetInventoryCycle) .where(AssetInventoryCycle.id == cycle_id) .with_for_update() ) ).scalar_one_or_none() if cycle_row is None: raise NotFoundError(f"inventory cycle {cycle_id} not found") cycle = cycle_row if cycle.status == "closed": return cycle if cycle.status != "active": raise ConflictError("only active cycles can be closed") cycle.status = "closed" cycle.closed_at = datetime.now(UTC) cycle.closed_by = actor_person_id # Settle the nag ladder for everyone this cycle was chasing. # Skipping this is the failure mode the cadence module's third # constraint names: the counters would survive, and the NEXT # cycle's very first reminder to a repeat participant would land # on the weekly rung — a noise fix quietly turning a brand-new # ask into near-silence. await cadence.clear( self.db, template_key=_REMINDER_TEMPLATE_KEY, subject_key=str(cycle_id), ) await self.db.flush() return cycle # ── Helpers used by scheduled job handlers ──────────────────────── async def _recompute_counters(self, cycle_id: UUID) -> None: """Recompute cycle counters from the items table. Single SQL pass over ``asset_inventory_items`` — atomic within the calling transaction, free of the read-modify-write race that bit the previous Python-level increment loop. Caller is expected to hold a row lock on the cycle to serialise across transactions. """ # One pass over ``asset_inventory_items`` filtered by cycle_id # — index ``ix_asset_inv_items_cycle_status`` covers it. counts_stmt = select( func.sum(case((AssetInventoryItem.status == "pending", 1), else_=0)), func.sum(case((AssetInventoryItem.status == "confirmed", 1), else_=0)), func.sum(case((AssetInventoryItem.status == "issue_reported", 1), else_=0)), func.count(), ).where(AssetInventoryItem.cycle_id == cycle_id) row = (await self.db.execute(counts_stmt)).one() pending, confirmed, issues, total = row cycle = await self.db.get(AssetInventoryCycle, cycle_id) if cycle is None: return cycle.pending_count = int(pending or 0) cycle.confirmed_count = int(confirmed or 0) cycle.issue_count = int(issues or 0) cycle.total_items = int(total or 0) await self.db.flush() async def cycles_due_for_reminder(self) -> list[AssetInventoryCycle]: """Active cycles past the 1-day grace window. Per-recipient daily dedup + local-working-hours gating is applied inside :meth:`send_cycle_reminder`; this query only filters at the cycle level so a launched-yesterday cycle starts producing daily nags from tomorrow morning onward. Active cycle count per deployment is expected ≤ a few dozen (one half-yearly + ad-hoc admin cycles), so an unfiltered scan + in-memory per-person fanout in the service is cheap. """ cutoff = datetime.now(UTC) - timedelta(days=_REMINDER_GRACE_DAYS) stmt = select(AssetInventoryCycle).where( AssetInventoryCycle.status == "active", AssetInventoryCycle.started_at.isnot(None), AssetInventoryCycle.started_at <= cutoff, ) rows = (await self.db.execute(stmt)).scalars().all() return list(rows) async def run_start_notifications_in_background( cycle_id: UUID, *, actor_user_id: UUID | None = None, trace_id: str = "", ) -> None: """Fan out an inventory cycle's launch notifications on a FRESH session, off the request thread — the coroutine the ``start_cycle`` / ``create_cycle(auto_start=True)`` HTTP paths hand to ``background.spawn`` after committing the (fast, all-DB) state transition and returning immediately. Owns its own DB session (the request's session is closed once the response is returned) and raises the per-session lock-wait ceiling the same way the HRIS background-apply path does, since a company-wide cycle chains hundreds of ``notification_outbox`` inserts. Best-effort: on a top-level crash the rows already enqueued in phase 1 are durably PENDING, so the hourly ``notification.outbox_retry`` worker delivers whatever this drain didn't reach — no card is silently lost. """ from sqlalchemy import text as _text from app.core.database import async_session_factory try: async with async_session_factory() as session: await session.execute( _text("SET SESSION innodb_lock_wait_timeout = 120") ) service = AssetInventoryService(session) sent = await service.dispatch_start_notifications( cycle_id, actor_user_id=actor_user_id, trace_id=trace_id, ) logger.info( "inventory start-notification fan-out finished for cycle %s: " "%s recipient(s)", cycle_id, sent, ) except Exception: # noqa: BLE001 logger.exception( "inventory start-notification fan-out crashed for cycle %s — " "rows enqueued so far stay PENDING and the " "notification.outbox_retry worker will deliver them.", cycle_id, )