Coverage for netbox_data_import/proposal_presentation.py: 98.62%
218 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"""Supply proposal cards and action reasons to the Review Workspace."""
5from urllib.parse import urlencode
7from core.choices import JobStatusChoices
8from django.core.exceptions import ValidationError
9from django.db.models import F, Window
10from django.db.models.functions import RowNumber
11from django.urls import reverse
12from django.utils.timesince import timesince
14from .field_keys import SELECT_TERMINATION_TASK, TERMINATION_ROLE, is_outside_claim, parse_termination_field_key
15from .inference_backend import (
16 NoActiveInferenceBackend,
17 proposal_candidate_limit,
18 proposal_eligible_set_limit,
19 resolve_active_backend,
20)
21from .inference_trust import InvalidInferenceConfiguration
22from .models import ImportProfile, ProposalDecision, ProposalOutcome, ProposalStatus, ResolutionProposal
23from .proposal_decisions import proposal_staleness
24from .review_workspace import OUTSIDE_CLAIM_MARK, TERMINATION_UNRESOLVABLE
25from .resolution_proposals import page_exhausted
26from .proposal_tasks import CandidateSnapshot, proposal_task
27from .cable_target import AUTOMATICALLY_RESOLVED, MANUALLY_RESOLVED, UNRESOLVED
29#: The badge modifier each field state wears, so the template needs no state vocabulary of its own.
30STATE_STYLES = {
31 "": "unknown",
32 UNRESOLVED: "unresolved",
33 AUTOMATICALLY_RESOLVED: "auto",
34 MANUALLY_RESOLVED: "manual",
35 "proposed": "proposed",
36 ProposalDecision.ACCEPTED: "accepted",
37 "stale": "stale",
38 ProposalStatus.FAILED: "failed",
39}
41RECENT_PROPOSAL_HISTORY_LIMIT = 10
42WAITING_FOR_SLOT = "Waiting for a free AI backend slot..."
43WAITING_FOR_BACKEND = "Waiting for the backend..."
44REQUEST_PERMISSION_REASON = "You do not have permission to request proposals for this Import Profile."
47def group_terminations(fields):
48 """Keep exact matches without proposal history in the compact settled group.
50 An end of a segment NetBox cannot cable stays in attention, so its picker can correct the match.
51 """
52 attention, settled = [], []
53 for field in fields:
54 settles = (
55 field["proposal"]["field_state"] == AUTOMATICALLY_RESOLVED
56 and not field["proposal_read"]["history_display"]
57 and not field["incompatible"]
58 )
59 (settled if settles else attention).append(field)
60 return attention, settled
63def backend_unavailable_reason() -> str:
64 """Return why no proposal can reach an Inference Backend now, or an empty string."""
65 try:
66 resolve_active_backend()
67 except NoActiveInferenceBackend:
68 return "No Inference Backend is enabled or configured as a fallback."
69 except (InvalidInferenceConfiguration, ValidationError):
70 return "The active Inference Backend configuration is invalid."
71 return ""
74#: The commands each card renders beside the termination picker; the others sit in the proposal display.
75FIELD_ACTIONS = frozenset({"request", "cancel"})
78def _action(key, label, reason, *, hidden, disabled=None):
79 """Return one command. A hidden command shows nothing; a disabled one states only its own refusal.
81 A failed or stale proposal refuses its decisions as a whole, and the card states that once.
82 """
83 return {
84 "key": key,
85 "label": label,
86 "reason": "" if hidden else reason,
87 "hidden": hidden,
88 "disabled": not hidden and (bool(reason) if disabled is None else disabled),
89 "url": reverse(f"plugins:netbox_data_import:trace_{key}_proposal"),
90 }
93class TerminationInventories:
94 """Read one inventory per Device, kind, and role, for the card display and for every request."""
96 def __init__(self, *, profile, reader):
97 self.profile = profile
98 self.reader = reader
99 self._reads = {}
101 def get(self, field_key):
102 """Return the cached inventory for the Device, kind, and role that *field_key* names."""
103 parsed = parse_termination_field_key(field_key)
104 key = (parsed["device"], parsed["kind"], parsed["role"])
105 if key not in self._reads:
106 self._reads[key] = proposal_task(SELECT_TERMINATION_TASK).inventory(
107 profile=self.profile,
108 field_key=field_key,
109 netbox_reader=self.reader,
110 limit=proposal_eligible_set_limit(),
111 )
112 return self._reads[key]
115class ProposalPresentation:
116 """Read one profile's proposal display with one backend lookup per response."""
118 def __init__(self, *, profile, actor, reader):
119 self.profile = profile
120 self.actor = actor
121 self.reader = reader
122 self._inventories = TerminationInventories(profile=profile, reader=reader)
123 self._write_assessments = {}
124 self.preview_allowed = ImportProfile.objects.restrict(actor, "change").filter(pk=profile.pk).exists()
125 self.profile_view_allowed = ImportProfile.objects.restrict(actor, "view").filter(pk=profile.pk).exists()
126 self.view_reason = ""
127 if not self.profile_view_allowed:
128 self.view_reason = "You do not have permission to view proposals for this Import Profile."
129 self.backend_reason = backend_unavailable_reason()
131 @property
132 def request_block_reason(self) -> str:
133 """Explain why this operator can ask about no field of the profile, or return an empty string."""
134 if self.view_reason:
135 return self.view_reason
136 if not self.preview_allowed:
137 return REQUEST_PERMISSION_REASON
138 return self.backend_reason
140 def fields(self, fields):
141 """Return each displayed field and its bounded history within the authorized profile."""
142 if self.view_reason:
143 return {field["field_key"]: self.field(field, None, [], False) for field in fields}
144 histories: dict[str, list] = {field["field_key"]: [] for field in fields}
145 rows = (
146 ResolutionProposal.objects.filter(
147 profile=self.profile, task_type=SELECT_TERMINATION_TASK, field_key__in=histories
148 )
149 .annotate(
150 history_position=Window(
151 expression=RowNumber(),
152 partition_by=F("field_key"),
153 order_by=(F("created").desc(), F("pk").desc()),
154 )
155 )
156 .filter(history_position__lte=RECENT_PROPOSAL_HISTORY_LIMIT + 1)
157 .only("pk", "field_key", "created", "status", "outcome", "decision", "failure_reason")
158 .order_by("field_key", "history_position")
159 )
160 for row in rows:
161 histories[row.field_key].append(row)
162 current_ids = [history[0].pk for history in histories.values() if history]
163 current = {
164 proposal.pk: proposal
165 for proposal in ResolutionProposal.objects.filter(pk__in=current_ids).select_related(
166 "profile", "resolved_device_type", "selected_object_type", "written_resolution", "job"
167 )
168 }
169 return {
170 field["field_key"]: self.field(
171 field,
172 current.get(history[0].pk) if history else None,
173 history[:RECENT_PROPOSAL_HISTORY_LIMIT],
174 len(history) > RECENT_PROPOSAL_HISTORY_LIMIT,
175 )
176 for field in fields
177 for history in (histories[field["field_key"]],)
178 }
180 def field(self, field, proposal, history, history_has_more):
181 """Serialize the current attempt, freshness, actions, and recent summaries."""
182 record = {"id": proposal.pk} if proposal is not None else None
183 history_url = None
184 if history:
185 query = urlencode({"profile_id": self.profile.pk, "field_key": field["field_key"]})
186 history_url = f"{reverse('plugins-api:netbox_data_import-api:resolutionproposalhistory-list')}?{query}"
187 payload = {
188 "proposal": record,
189 "history_display": [
190 {
191 "id": row.pk,
192 "created": row.created.isoformat(),
193 "status": row.get_status_display(),
194 "outcome": row.get_outcome_display() or "No outcome",
195 "decision": row.get_decision_display(),
196 "failure": row.get_failure_reason_display(),
197 }
198 for row in history
199 ],
200 "history_has_more": history_has_more,
201 "history_url": history_url,
202 "staleness": None,
203 }
204 inventory = self.field_inventory(field) if proposal is not None or self.preview_allowed else None
205 if self.reader is None:
206 payload["staleness_error"] = "The saved import target is gone or outside your view scope."
207 elif proposal is not None:
208 stale = proposal_staleness(proposal, inventory=inventory)
209 payload["staleness"] = {
210 "is_stale": stale.is_stale,
211 "resolved_device_changed": stale.resolved_device_changed,
212 "candidates_changed": stale.candidates_changed,
213 }
214 payload["presentation"] = self.card(field, proposal, payload, inventory)
215 return payload
217 def field_inventory(self, field):
218 """Return one cached inventory read for fields that share device, kind, and role."""
219 if self.reader is None or parse_termination_field_key(field["field_key"])["role"] != TERMINATION_ROLE:
220 return None
221 return self._inventories.get(field["field_key"])
223 def action_permission_reason(self, field, inventory):
224 """Explain access shared by request and cancellation actions."""
225 if not self.preview_allowed:
226 return REQUEST_PERMISSION_REASON
227 if parse_termination_field_key(field["field_key"])["role"] != TERMINATION_ROLE:
228 return "Ask AI supports termination fields only. Choose the mapped peer manually."
229 if inventory is None or inventory.resolved_device is None:
230 return "The resolved Device is unavailable or outside your view permission."
231 return ""
233 def request_permission_reason(self, field, inventory):
234 """Explain access and candidate eligibility for a proposal request."""
235 if field.get("source_ambiguous", False):
236 return TERMINATION_UNRESOLVABLE
237 reason = self.action_permission_reason(field, inventory)
238 if reason:
239 return reason
240 if inventory.candidate_error is not None:
241 return str(inventory.candidate_error)
242 return ""
244 def card(self, field, proposal, payload, inventory):
245 """Derive all proposal vocabulary and legal actions in one place."""
246 pending = proposal is not None and proposal.status in ProposalStatus.ACTIVE
247 # A queued row without a Job waits for the dispatcher to free a request slot.
248 waiting = pending and proposal.status == ProposalStatus.QUEUED and proposal.job_id is None
249 completed = proposal is not None and proposal.status == ProposalStatus.COMPLETED
250 state = field["state"]
251 # The plan owns whether a field is resolved; acceptance only renames a state it already set.
252 if state != UNRESOLVED and proposal is not None and proposal.decision == ProposalDecision.ACCEPTED:
253 resolution = proposal.written_resolution
254 if resolution is not None and (
255 resolution.selected_object_type_id == proposal.selected_object_type_id
256 and resolution.selected_object_id == proposal.selected_object_id
257 ):
258 state = ProposalDecision.ACCEPTED
259 stale = payload["staleness"]
260 stale_reason = payload.get("staleness_error", "")
261 if stale and stale["is_stale"]:
262 stale_reason = "The resolved Device or eligible candidates changed. Request a new proposal."
263 candidate, missing, selected_entry = ("", False, None) if stale_reason else self.selected_candidate(proposal)
264 if missing:
265 # Acceptance refuses this row, so the card must not offer an action the writer declines.
266 stale_reason = "The selected candidate is no longer in the request snapshot. Request a new proposal."
267 offered = self.offered_page(proposal)
268 actions = self.actions(
269 field,
270 proposal,
271 state,
272 pending,
273 completed,
274 stale_reason,
275 selected_entry,
276 inventory,
277 offered,
278 )
279 badge = proposal.get_status_display() if proposal is not None else "No proposal"
280 if completed:
281 badge = "Proposal - stale, not applied" if stale_reason else "Proposal - not applied"
282 if proposal is not None and proposal.decision:
283 badge = proposal.get_decision_display()
284 if proposal.decision == ProposalDecision.ACCEPTED and state == UNRESOLVED:
285 badge = "Accepted resolution no longer applies"
286 if state == UNRESOLVED and proposal is not None and not proposal.decision:
287 if pending or proposal.outcome == ProposalOutcome.CANDIDATE:
288 state = "proposed"
289 if completed and stale_reason:
290 state = "stale"
291 if proposal.status == ProposalStatus.FAILED:
292 state = ProposalStatus.FAILED
293 return {
294 "has_proposal": proposal is not None,
295 "pending": pending,
296 "progress": WAITING_FOR_SLOT if waiting else WAITING_FOR_BACKEND,
297 "field_state": state,
298 "state_style": STATE_STYLES[state],
299 "badge": badge,
300 "candidate": candidate,
301 # A proposal is requested for the termination role only, so its field key states the source claim.
302 "outside_claim": OUTSIDE_CLAIM_MARK
303 if selected_entry is not None
304 and is_outside_claim(parse_termination_field_key(field["field_key"])["kind"], selected_entry.object_type)
305 else "",
306 "explanation": proposal.explanation if proposal is not None else "",
307 "failure": proposal.get_failure_reason_display() if proposal is not None else "",
308 "failure_code": proposal.failure_reason if proposal is not None else "",
309 "job_status": self.job_status(proposal) if pending else "",
310 "job_note": self.job_note(proposal) if pending else "",
311 "page_status": self.page_status(offered) if completed else "",
312 "no_match": completed and proposal.outcome == ProposalOutcome.NO_MATCH,
313 # Accept can be hidden, so the card itself warns that an undecided answer no longer applies.
314 "warning": stale_reason if completed and not proposal.decision else "",
315 "actions": actions,
316 "field_actions": [action for action in actions if action["key"] in FIELD_ACTIONS and not action["hidden"]],
317 "decision_actions": [
318 action for action in actions if action["key"] not in FIELD_ACTIONS and not action["hidden"]
319 ],
320 }
322 @staticmethod
323 def offered_page(proposal):
324 """Return the page one attempt offered, or None when it offered the whole eligible set."""
325 if proposal is None:
326 return None
327 snapshot = CandidateSnapshot.from_json(proposal.candidate_snapshot)
328 return snapshot if snapshot.total > len(snapshot.page) else None
330 @staticmethod
331 def page_status(offered):
332 """Name the candidates one attempt searched, which is what a no_match is bounded by."""
333 if offered is None:
334 return ""
335 return f"Searched candidates {offered.page_offset + 1}-{offered.page_end} of {offered.total}."
337 @staticmethod
338 def job_status(proposal):
339 """Say where the background job stands, which a card waiting on one cannot otherwise show."""
340 job = proposal.job
341 if job is not None:
342 state = job.get_status_display()
343 else:
344 state = "not started yet" if proposal.status == ProposalStatus.QUEUED else "none recorded"
345 return f"Background job: {state}, requested {timesince(proposal.created)} ago"
347 @staticmethod
348 def job_note(proposal):
349 """Name the two states in which an active attempt is never going to get an answer."""
350 job = proposal.job
351 if job is None and proposal.status == ProposalStatus.QUEUED:
352 return ""
353 if job is None:
354 return "No background job is recorded. Cancel this proposal and ask again."
355 if job.status in JobStatusChoices.TERMINAL_STATE_CHOICES:
356 return "The background job ended without recording a result. Cancel this proposal and ask again."
357 return ""
359 @staticmethod
360 def selected_candidate(proposal):
361 """Return the chosen candidate's display, and whether the snapshot no longer holds it."""
362 if proposal is None or proposal.outcome != ProposalOutcome.CANDIDATE:
363 return "", False, None
364 snapshot = CandidateSnapshot.from_json(proposal.candidate_snapshot)
365 entry = next(
366 (row for row in snapshot.entries if row.candidate_id == proposal.selected_candidate_id),
367 None,
368 )
369 if entry is None:
370 return "", True, None
371 return f"{entry.display_name} ({str(proposal.selected_object_type.name).capitalize()})", False, entry
373 def actions(self, field, proposal, state, pending, completed, stale_reason, selected_entry, inventory, offered):
374 """Return every command with its own refusal, hiding a command the lifecycle rules out."""
375 failed = proposal is not None and proposal.status == ProposalStatus.FAILED
376 decided = proposal is not None and bool(proposal.decision)
377 decision_hidden = decided or not (completed or failed)
378 no_match = completed and proposal.outcome == ProposalOutcome.NO_MATCH
379 accept_reason = TERMINATION_UNRESOLVABLE if field.get("source_ambiguous", False) else ""
380 write_refused = not self.preview_allowed or (
381 selected_entry is not None and "source" in field and not self.write_allowed(field, proposal, selected_entry)
382 )
383 if write_refused:
384 accept_reason = "You do not have permission to save a termination resolution."
385 reject_reason = "" if self.profile_view_allowed else "You do not have permission to reject proposals."
386 actions = [
387 _action(
388 "request",
389 self.request_label(proposal, offered, stale_reason),
390 self.request_permission_reason(field, inventory) or self.backend_reason,
391 hidden=state != UNRESOLVED or pending,
392 ),
393 _action("cancel", "Cancel", self.action_permission_reason(field, inventory), hidden=not pending),
394 _action(
395 "accept",
396 "Accept",
397 accept_reason,
398 hidden=decision_hidden or no_match,
399 disabled=failed or bool(stale_reason) or bool(accept_reason),
400 ),
401 _action("reject", "Reject", reject_reason, hidden=decision_hidden, disabled=failed or bool(reject_reason)),
402 ]
403 if self.view_reason:
404 # Without view access the card shows no proposal, so it states the refusal on every command.
405 return [{**action, "reason": self.view_reason, "hidden": False, "disabled": True} for action in actions]
406 return actions
408 def write_allowed(self, field, proposal, selected_entry) -> bool:
409 """Return whether this actor may write the resolution the selected candidate names, read once per write."""
410 assessment_key = (
411 proposal.field_key,
412 tuple(field["source"][part] for part in ("device", "cards", "port")),
413 selected_entry.object_type,
414 selected_entry.object_id,
415 selected_entry.display_name,
416 )
417 if assessment_key not in self._write_assessments:
418 self._write_assessments[assessment_key] = proposal_task(SELECT_TERMINATION_TASK).assess_resolution_write(
419 profile=self.profile,
420 field_key=proposal.field_key,
421 entry=selected_entry,
422 source=field["source"],
423 actor=self.actor,
424 )
425 return self._write_assessments[assessment_key].allowed
427 @staticmethod
428 def next_page_size(offered) -> int:
429 """Return how many candidates the next request offers, which is one page, not the rest."""
430 if offered is None or not offered.has_next_page:
431 return 0
432 return min(offered.total - offered.page_end, proposal_candidate_limit())
434 @classmethod
435 def request_label(cls, proposal, offered, stale_reason) -> str:
436 """Name what the next request does: retry, continue the search, or start one."""
437 if proposal is not None and proposal.status == ProposalStatus.FAILED:
438 return "Ask AI again"
439 if proposal is None or stale_reason or not page_exhausted(proposal):
440 return "Ask AI"
441 if offering := cls.next_page_size(offered):
442 return f"Ask AI: next {offering}"
443 return "Ask AI"