Files
compass_backoffice/src/api/utils/emulator_response_builder.py
2026-06-13 08:23:21 -03:00

203 lines
8.5 KiB
Python

"""Builds the `TicketResponseEvent` for the Response Emulator flow.
Kept separate from `agent_helpers.build_cms_response_event` (checklist's
builder) to avoid coupling the two domains. Status mapping (status,action):
error at any step → failed, retry
flow_mode="generate" → awaiting_review, await_response
flow_mode="approve" → approved, await_response
flow_mode="close" → done, response
The event is meant to be the LAST message published on the OCI Response
Stream for a given case, so the CMS callback resolves the final DB state
without races against earlier ProgressEvents.
## Where each field lives (split by lifetime)
- `processing.metadata` (per-run scratch, lives only in the latest
`processing` subdoc): just `error` when the graph failed. Everything else
used to be here too and was duplicating the case-doc `metadata` below —
the GET endpoint already reads validation/selected_actions from the
case-level `metadata`, so keeping a parallel copy under `processing` was
pure dead weight that the CMS overwrote on every run.
- `metadata` (top-level case doc, persisted incrementally via `$set` with
dot-notation): selected_actions, validation, flow_mode, emulator_routing,
and the `last_emulation` summary. These survive across runs and merge
cleanly with whatever the checklist already wrote to `metadata`.
"""
from datetime import datetime, timezone
from src.agent.state.agent_state import AgentState
from src.agent.state.steps_emulator import EmulatorGraphStep
from src.api.schemas.anatel_schemas import (
Processing,
ProcessingAction,
ProcessingStatus,
TicketResponseEvent,
)
# Maps flow_mode → (status, action) when the graph completes without error.
_FLOW_MODE_OUTCOMES: dict[str, tuple[ProcessingStatus, ProcessingAction]] = {
"generate": (ProcessingStatus.AWAITING_REVIEW, ProcessingAction.AWAIT_RESPONSE),
"approve": (ProcessingStatus.APPROVED, ProcessingAction.AWAIT_RESPONSE),
"close": (ProcessingStatus.DONE, ProcessingAction.RESPONSE),
}
def _resolve_crm_protocol(case_data: dict | None) -> str | None:
"""Returns the persisted Siebel SR protocol, or None when no SR exists.
Why this is restricted to two sources (and NOT `case_data.crmProtocol`
at the root or `request_context.crmProtocol`): the root-level field on
the request event is whatever the simulator/CMS sent in — for our test
payloads it's `"DS-XXXXXXXX"`, a ticketId-shaped placeholder, not the
real Anatel-style SR protocol (`"20260..."`) that Siebel returns when
the treatment SR opens. Falling back to it would propagate that bogus
value into `processing.crmProtocol` and the close PATCH would close
the wrong SR.
Trustworthy sources, in order:
1. `processing.crmProtocol` — written by the checklist's terminal
event from `siebel_sr_data.interactionProtocol`. Canonical.
2. `siebel_sr_data.interactionProtocol` — in-memory only; relevant
when the checklist + emulator run in the same process (tests).
The emulator MUST echo this back on every terminal event because the
CMS callback overwrites the `processing` subdoc; omitting the field
zeroes it, which is the bug that broke `close_case` before.
"""
case_data = case_data or {}
processing = case_data.get("processing") or {}
sr_data = case_data.get("siebel_sr_data") or {}
return processing.get("crmProtocol") or sr_data.get("interactionProtocol")
def _resolve_persisted_field(
case_data: dict | None,
new_value,
field: str,
):
"""Returns `new_value` when truthy, else the persisted value in `case_data.processing.{field}`.
Why: when a graph run fails (e.g. `approve_draft` errors because of a
transient issue), the state's `metadata.{field}` is empty for that
run. If we publish `Processing(field=None)`, the CMS callback wipes
the value the previous successful run had persisted — turning a
recoverable failure into permanent data loss (this is exactly what
nuked `case_response`/`transitions` mid-flow before this fix). When
the new run actually produced a value, that wins; otherwise we echo
what is already in the doc so the failure is non-destructive.
"""
if new_value:
return new_value
processing = (case_data or {}).get("processing") or {}
return processing.get(field)
def _resolve_outcome(
error_info: dict | None,
case_response: str | None,
flow_mode: str | None,
) -> tuple[ProcessingStatus, ProcessingAction]:
if error_info and flow_mode == "close":
return ProcessingStatus.SIEBEL_CLOSING_FAILED, ProcessingAction.RETRY
if error_info:
return ProcessingStatus.FAILED, ProcessingAction.RETRY
if flow_mode == "close" and not case_response:
return ProcessingStatus.FAILED, ProcessingAction.RETRY
if flow_mode == "generate" and not case_response:
return ProcessingStatus.FAILED, ProcessingAction.RETRY
return _FLOW_MODE_OUTCOMES.get(
flow_mode or "",
(ProcessingStatus.FAILED, ProcessingAction.RETRY),
)
def _build_case_metadata(
selected_actions: list,
validation: dict,
flow_mode: str | None,
routing: dict | None,
emulation_type: str | None,
iteration_count: int,
) -> dict:
"""Top-level case-doc metadata for the CMS to `$set` via dot notation.
Empty/None values are omitted so a partial run (e.g. `approve` has no
new `selected_actions` or `validation`) doesn't blow away keys written
by an earlier run — this is the incremental-merge contract with the
CMS callback. `last_emulation` is always included as a fresh marker
of when the most recent graph run happened and what it was.
"""
case_metadata: dict = {
"last_emulation": {
"type": emulation_type,
"is_regeneration": emulation_type == "regenerate",
"flow_mode": flow_mode,
"iteration_count": iteration_count,
"at": datetime.now(timezone.utc).isoformat(),
},
}
if selected_actions:
case_metadata["selected_actions"] = selected_actions
if validation:
case_metadata["validation"] = validation
if flow_mode:
case_metadata["flow_mode"] = flow_mode
if routing:
case_metadata["emulator_routing"] = routing
return case_metadata
def build_emulator_response_event(state: AgentState, transaction_id: str) -> TicketResponseEvent:
metadata = state.get("metadata", {}) or {}
request_context = metadata.get("request_context") or {}
case_response = metadata.get("case_response") or request_context.get("case_response")
current_step = str(state.get("current_step") or "")
error_info = state.get("error")
validation = metadata.get("validation") or {}
flow_mode = metadata.get("flow_mode") or request_context.get("flow_mode")
transitions = metadata.get("transitions")
selected_actions = metadata.get("selected_actions") or []
routing = metadata.get("emulator_routing")
emulation_type = request_context.get("type")
iteration_count = state.get("iteration_count", 0)
case_data = metadata.get("case_data") or {}
crm_protocol = _resolve_crm_protocol(case_data)
# Echo persisted values when this run didn't produce new ones so a
# failed run doesn't blank out the previous successful draft.
case_response = _resolve_persisted_field(case_data, case_response, "case_response")
transitions = _resolve_persisted_field(case_data, transitions, "transitions")
cms_status, action = _resolve_outcome(error_info, case_response, flow_mode)
# Per-run scratch only — case-level info goes under top-level `metadata`.
response_metadata: dict = {}
if error_info:
response_metadata["error"] = error_info
case_metadata = _build_case_metadata(
selected_actions=selected_actions,
validation=validation,
flow_mode=flow_mode,
routing=routing,
emulation_type=emulation_type,
iteration_count=iteration_count,
)
return TicketResponseEvent(
transactionId=transaction_id,
processing=Processing(
status=cms_status,
current_step=current_step or str(EmulatorGraphStep.CASE_CLOSED),
action=action,
note=state.get("processing_notes") or None,
crmProtocol=crm_protocol,
case_response=case_response,
transitions=transitions,
metadata=response_metadata or None,
),
metadata=case_metadata,
)