Coverage for netbox_data_import/resolution_proposals.py: 98.63%
73 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Resolution Proposal lifecycle: request, claim, complete, fail, cancel, decide (section 7.2).
5Every ordering-dependent rule is a conditional `UPDATE` whose rowcount is the refusal. A read followed
6by a write would let a cancellation and a late worker response both believe they won.
7"""
9from django.db import IntegrityError, transaction
10from django.utils import timezone
12from .models import (
13 ProposalDecision,
14 ProposalFailureReason,
15 ProposalOutcome,
16 ProposalStatus,
17 ResolutionProposal,
18 index_digest,
19)
20from .proposal_tasks import CandidateSnapshot, proposal_inventory_staleness
21from .public_refusal import PublicRefusal
23__all__ = [
24 "ActiveProposalExists",
25 "active_proposal_exists",
26 "cancel_proposal",
27 "claim_proposal",
28 "complete_proposal",
29 "decide_proposal",
30 "fail_abandoned_proposal",
31 "fail_proposal",
32 "fail_queued_proposal",
33 "latest_proposal",
34 "next_page_offset",
35 "page_exhausted",
36 "record_proposal_job",
37 "request_proposal",
38]
41class ActiveProposalExists(PublicRefusal):
42 """One key already has a queued or running proposal, which the partial unique index refuses."""
45def request_proposal(
46 *,
47 profile,
48 task_type,
49 field_key,
50 source_evidence,
51 resolved_device_type,
52 resolved_device_id,
53 prompt_version,
54 response_schema_version,
55 candidate_snapshot: CandidateSnapshot,
56 requested_by=None,
57) -> ResolutionProposal:
58 """Create the queued row that is also the attempt record, before any backend call."""
59 if not isinstance(candidate_snapshot, CandidateSnapshot):
60 raise TypeError("candidate_snapshot must be a CandidateSnapshot.")
61 proposal = ResolutionProposal(
62 profile=profile,
63 task_type=task_type,
64 field_key=field_key,
65 status=ProposalStatus.QUEUED,
66 source_evidence=source_evidence,
67 resolved_device_type=resolved_device_type,
68 resolved_device_id=resolved_device_id,
69 prompt_version=prompt_version,
70 response_schema_version=response_schema_version,
71 candidate_snapshot=candidate_snapshot.as_json(),
72 requested_by=requested_by,
73 )
74 try:
75 with transaction.atomic():
76 proposal.save()
77 except IntegrityError as exc:
78 if "ndi_resolutionproposal_one_active" not in str(exc):
79 raise
80 raise ActiveProposalExists("This field already has an active Resolution Proposal.") from exc
81 return proposal
84def active_proposal_exists(*, profile, task_type, field_key) -> bool:
85 """Return whether this field already has a queued or running attempt.
87 The partial unique index refuses two active rows, but it reads the world at insert time. A
88 worker that settles between the page read and the insert would let the next request repeat a
89 page, so the request refuses on what it observed.
90 """
91 return ResolutionProposal.objects.filter(
92 profile=profile,
93 task_type=task_type,
94 field_key=field_key,
95 field_key_digest=index_digest(field_key),
96 status__in=ProposalStatus.ACTIVE,
97 ).exists()
100def page_exhausted(proposal) -> bool:
101 """Return whether one settled attempt used up the page it offered.
103 A `no_match` found nothing there, and a rejected candidate was the wrong answer from there.
104 Both leave the rest of the eligible set unsearched.
105 """
106 return proposal.outcome == ProposalOutcome.NO_MATCH or proposal.decision == ProposalDecision.REJECTED
109def latest_proposal(*, profile, task_type, field_key):
110 """Return the most recent attempt for one field, or None."""
111 return (
112 ResolutionProposal.objects.filter(profile=profile, task_type=task_type, field_key=field_key)
113 .order_by("-created", "-pk")
114 .first()
115 )
118def next_page_offset(*, profile, task_type, field_key, inventory) -> int:
119 """Return where the next request starts: after the last page that used itself up.
121 Any other last attempt leaves its page unfinished, so the request asks that page again.
122 Evidence that has moved since, or a used-up page that reached the end, starts the search again
123 at the first candidate. Freshness is the same read the card shows, so the offer and the request
124 cannot disagree: a replaced Device renumbers every page just as a changed candidate set does,
125 even when the ports themselves moved across unchanged.
126 """
127 previous = latest_proposal(profile=profile, task_type=task_type, field_key=field_key)
128 if previous is None or proposal_inventory_staleness(previous, inventory).is_stale:
129 return 0
130 offered = CandidateSnapshot.from_json(previous.candidate_snapshot)
131 if not page_exhausted(previous):
132 return offered.page_offset
133 return offered.page_end if offered.has_next_page else 0
136def record_proposal_job(proposal_id, job) -> None:
137 """Bind the enqueued Job to its attempt, in the dispatch transaction before the queue push."""
138 ResolutionProposal.objects.filter(pk=proposal_id).update(job=job)
141def _transition(proposal_id, *, allowed_from, **values) -> bool:
142 """Move one row between statuses, returning whether this caller is the one that moved it."""
143 values["last_updated"] = timezone.now()
144 moved = ResolutionProposal.objects.filter(pk=proposal_id, status__in=allowed_from).update(**values)
145 return moved == 1
148def claim_proposal(proposal_id) -> bool:
149 """Take a queued row for one worker. A second worker gets False and must not call the backend."""
150 return _transition(proposal_id, allowed_from=(ProposalStatus.QUEUED,), status=ProposalStatus.RUNNING)
153def _validate_choice(value, choices, *, name) -> None:
154 """Reject a value outside one persisted lifecycle vocabulary."""
155 if not any(value == allowed for allowed, _label in choices):
156 raise ValueError(f"Unsupported {name}.")
159def complete_proposal(
160 proposal_id,
161 *,
162 outcome,
163 explanation,
164 selected_candidate_id="",
165 selected_object_type=None,
166 selected_object_id=None,
167 backend_metadata=None,
168 response_diagnostic=None,
169) -> bool:
170 """Record the outcome of a running row. False means it was cancelled or already terminal."""
171 _validate_choice(outcome, ProposalOutcome.CHOICES, name="proposal outcome")
172 if outcome == ProposalOutcome.CANDIDATE and not selected_candidate_id:
173 raise ValueError("A candidate outcome requires selected_candidate_id.")
174 if outcome == ProposalOutcome.NO_MATCH:
175 selected_candidate_id, selected_object_type, selected_object_id = "", None, None
176 return _transition(
177 proposal_id,
178 allowed_from=(ProposalStatus.RUNNING,),
179 status=ProposalStatus.COMPLETED,
180 outcome=outcome,
181 explanation=explanation,
182 selected_candidate_id=selected_candidate_id,
183 selected_object_type=selected_object_type,
184 selected_object_id=selected_object_id,
185 backend_metadata=backend_metadata,
186 response_diagnostic=response_diagnostic,
187 )
190def fail_proposal(proposal_id, *, reason, response_diagnostic=None, backend_metadata=None) -> bool:
191 """Fail a queued or running row with its typed reason and whatever the call received."""
192 _validate_choice(reason, ProposalFailureReason.CHOICES, name="proposal failure reason")
193 return _transition(
194 proposal_id,
195 allowed_from=ProposalStatus.ACTIVE,
196 status=ProposalStatus.FAILED,
197 failure_reason=reason,
198 response_diagnostic=response_diagnostic,
199 backend_metadata=backend_metadata,
200 )
203def fail_abandoned_proposal(proposal_id, *, job) -> bool:
204 """Fail a running row that an earlier delivery of *job* claimed and never finished."""
205 moved = ResolutionProposal.objects.filter(pk=proposal_id, status=ProposalStatus.RUNNING, job=job).update(
206 status=ProposalStatus.FAILED,
207 failure_reason=ProposalFailureReason.QUEUE_UNAVAILABLE,
208 last_updated=timezone.now(),
209 )
210 return moved == 1
213def fail_queued_proposal(proposal_id, *, reason) -> bool:
214 """Fail a row no worker has taken yet. A running row belongs to its worker, whose answer must still land."""
215 _validate_choice(reason, ProposalFailureReason.CHOICES, name="proposal failure reason")
216 return _transition(
217 proposal_id, allowed_from=(ProposalStatus.QUEUED,), status=ProposalStatus.FAILED, failure_reason=reason
218 )
221def cancel_proposal(proposal_id) -> bool:
222 """Cancel a queued or running row. A response arriving afterwards cannot move it again."""
223 return _transition(proposal_id, allowed_from=ProposalStatus.ACTIVE, status=ProposalStatus.CANCELLED)
226def decide_proposal(proposal_id, *, decision, operator, written_resolution=None) -> bool:
227 """Set the one-shot decision fields, which never change the status.
229 Only a completed row that nobody has decided yet qualifies, so the rowcount also refuses a second
230 decision racing the first.
231 """
232 _validate_choice(decision, ProposalDecision.CHOICES, name="proposal decision")
233 if operator is None:
234 raise ValueError("A decision requires an operator.")
235 if decision == ProposalDecision.ACCEPTED and written_resolution is None:
236 raise ValueError("An accepted proposal requires a written resolution.")
237 values = {
238 "decision": decision,
239 "decided_by": operator,
240 "decided_at": timezone.now(),
241 "last_updated": timezone.now(),
242 }
243 if decision == ProposalDecision.ACCEPTED:
244 values["written_resolution"] = written_resolution
245 decided = ResolutionProposal.objects.filter(pk=proposal_id, status=ProposalStatus.COMPLETED, decision="").update(
246 **values
247 )
248 return decided == 1