"""lfcs-override-capture@v1 — typed-fallback path. Per Stage 5 milestone 1 + user override 2026-05-07: voice capture deferred (Task 0 spike with Rocky + Hus pending). Typed-fallback is the working v1. Schema designed so voice can be added later without rework — this module already accepts a `verbal_*` payload shape; the voice path can fill those fields without changing the Airtable writes. Usage (programmatic): from lib.override_capture import OverrideCaptureSession with OverrideCaptureSession( bid_record_id="recXXXXXXXXXXXXXX", agent_id="claude-code-opsman", participants=["Rocky", "Liam"], ) as session: session.capture_override( bid_line_item_id="recYYYYYYYYYYYYYY", original_rate_ex_gst=37500.0, override_rate_ex_gst=30000.0, reason_codes=["PRODUCTIVITY-CALIBRATION"], reviewer_type="Trades-Internal-Liam", reviewer_name="Liam", free_text_rationale="Regular rectangle vs irregular pentagon assumption.", ) Usage (CLI typed prompt loop) — see `scripts/cli_capture.py` if/when needed. """ from __future__ import annotations import datetime as dt import json import os import uuid from pathlib import Path from typing import Any from . import airtable_client as at from .config import BASE_ID, FIELD, REASON_CODE_DOMAINS, REASON_CODES, TBL VAULT_INTERNAL_ROOT = Path( os.environ.get( "LFCS_VAULT_ROOT", r"C:\Users\mclou\rateright-growth-deploy\HQ-Vault\11_OpsMan_LFCS", ) ) / "_Internal" # Reviewer types accepted by Bid Line Item Overrides.reviewer_type singleSelect ALLOWED_REVIEWER_TYPES = { "Trades-Internal-Rocky", "Trades-Internal-Liam", "Trades-Internal-Hasibul", "Commercial-External-Robert", "Commercial-External-CivilExperts", "Commercial-External-Other", "Post-Submission-Outcome", } def _now_iso() -> str: return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") def _validate_reason_codes(codes: list[str]) -> None: bad = [c for c in codes if c not in REASON_CODE_DOMAINS] if bad: raise ValueError( f"Unknown reason codes {bad}. Accepted: {sorted(REASON_CODE_DOMAINS)}." ) class OverrideCaptureSession: """Context manager for one sanity-check session. On enter: opens an `_Internal_Session_Logs` row (capture_mode=`typed` until voice ships). On capture_override: writes `Bid_Line_Item_Overrides` + `_Internal_Audit_Trail`, appends to per-job `_audit/overrides-log-{date}.md` + cross-job `_Internal/audit-trail/session-{uuid}.md` (vault sources of truth). On exit: stamps `ended_at` on the session log + writes session summary. All writes are best-effort. If Airtable is unreachable, captures land in a `_Internal/audit-trail/pending-overrides-{session}.json` queue for later flush (matches Stage 4 § 3.4.3 fallback). The vault md files always write. """ def __init__( self, bid_record_id: str, agent_id: str, participants: list[str] | None = None, capture_mode: str = "typed", per_job_audit_dir: Path | None = None, offline: bool = False, ): self.bid_record_id = bid_record_id self.agent_id = agent_id self.participants = participants or [] self.capture_mode = capture_mode self.session_ref = str(uuid.uuid4()) self.session_record_id: str | None = None self.captures: list[dict] = [] self.offline = offline self.per_job_audit_dir = per_job_audit_dir self._pending_path = ( VAULT_INTERNAL_ROOT / "audit-trail" / f"pending-overrides-{self.session_ref}.json" ) # ---- context manager ---- def __enter__(self) -> "OverrideCaptureSession": VAULT_INTERNAL_ROOT.mkdir(parents=True, exist_ok=True) (VAULT_INTERNAL_ROOT / "audit-trail").mkdir(parents=True, exist_ok=True) (VAULT_INTERNAL_ROOT / "prompt-logs").mkdir(parents=True, exist_ok=True) if not self.offline: try: self._open_session_log() except Exception as e: # noqa: BLE001 # Degrade to offline-queue mode rather than aborting the session. self.offline = True self._log(f"WARN: Airtable session-log open failed ({e}); switching to offline.") self._log( f"SESSION_STARTED session_ref={self.session_ref} bid={self.bid_record_id} " f"agent={self.agent_id} mode={self.capture_mode} offline={self.offline}" ) return self def __exit__(self, exc_type, exc, tb) -> None: self._log(f"SESSION_ENDED captures={len(self.captures)}") if not self.offline and self.session_record_id: try: at.update_records( BASE_ID, TBL["_Internal_Session_Logs"], [ { "id": self.session_record_id, "fields": { FIELD["_Internal_Session_Logs"]["ended_at"]: _now_iso(), FIELD["_Internal_Session_Logs"]["transcript_summary"]: ( self._summary() ), }, } ], ) except Exception as e: # noqa: BLE001 self._log(f"WARN: session-log close failed ({e}).") # ---- public API ---- def capture_override( self, bid_line_item_id: str, original_rate_ex_gst: float, override_rate_ex_gst: float, reason_codes: list[str], reviewer_type: str, reviewer_name: str, free_text_rationale: str, original_qty: float | None = None, override_qty: float | None = None, verbal_capture: bool = False, verbal_transcript_excerpt: str | None = None, agent_first_pass_confidence: float | None = None, linked_methodology_pattern_ids: list[str] | None = None, linked_rate_template_ids: list[str] | None = None, ) -> dict: """Validate + commit one override row. Returns the captured payload (with `id`).""" _validate_reason_codes(reason_codes) if reviewer_type not in ALLOWED_REVIEWER_TYPES: raise ValueError(f"reviewer_type {reviewer_type!r} not in {ALLOWED_REVIEWER_TYPES}.") if not reviewer_name.strip(): raise ValueError("reviewer_name required.") if not free_text_rationale.strip(): raise ValueError("free_text_rationale required (Stage 4 § 3.1).") capture_id = str(uuid.uuid4()) f = FIELD["Bid_Line_Item_Overrides"] fields_payload = { f["id"]: capture_id, f["bid_line_item"]: [bid_line_item_id], f["original_rate_ex_gst"]: float(original_rate_ex_gst), f["override_rate_ex_gst"]: float(override_rate_ex_gst), f["reason_codes"]: list(reason_codes), f["reviewer_type"]: reviewer_type, f["reviewer_name"]: reviewer_name, f["free_text_rationale"]: free_text_rationale, f["verbal_capture"]: bool(verbal_capture), f["created_at"]: _now_iso(), f["session_ref"]: self.session_ref, } if original_qty is not None: fields_payload[f["original_qty"]] = float(original_qty) if override_qty is not None: fields_payload[f["override_qty"]] = float(override_qty) if verbal_transcript_excerpt: fields_payload[f["verbal_transcript_excerpt"]] = verbal_transcript_excerpt if agent_first_pass_confidence is not None: fields_payload[f["_internal_agent_first_pass_confidence"]] = float( agent_first_pass_confidence ) if self.session_record_id: fields_payload[f["linked_session_log"]] = [self.session_record_id] if linked_methodology_pattern_ids: fields_payload[f["linked_methodology_pattern"]] = list(linked_methodology_pattern_ids) if linked_rate_template_ids: fields_payload[f["linked_rate_template"]] = list(linked_rate_template_ids) record_id = self._commit_override(fields_payload) capture_record = {"id": record_id, "uuid": capture_id, "fields": fields_payload} self.captures.append(capture_record) self._mirror_to_vault(capture_record) self._emit_audit_event( event_type="override_captured", linked_table="Bid_Line_Item_Overrides", linked_id=record_id or capture_id, confidence=agent_first_pass_confidence, notes=f"reason_codes={reason_codes} reviewer={reviewer_type}/{reviewer_name}", ) return capture_record # ---- internal ---- def _open_session_log(self) -> None: f = FIELD["_Internal_Session_Logs"] payload = { f["id"]: self.session_ref, # use the session UUID as the id field f["session_ref"]: self.session_ref, f["agent_id"]: self.agent_id, f["started_at"]: _now_iso(), f["participant_names"]: ", ".join(self.participants) if self.participants else "", f["linked_bid"]: [self.bid_record_id], f["capture_mode"]: self.capture_mode, } records = at.create_records(BASE_ID, TBL["_Internal_Session_Logs"], [payload]) if records: self.session_record_id = records[0]["id"] self._emit_audit_event( event_type="agent_session_started", linked_table="_Internal_Session_Logs", linked_id=self.session_record_id, ) def _commit_override(self, fields_payload: dict) -> str | None: if self.offline: self._enqueue_pending(fields_payload) return None try: recs = at.create_records(BASE_ID, TBL["Bid_Line_Item_Overrides"], [fields_payload]) return recs[0]["id"] if recs else None except Exception as e: # noqa: BLE001 self._log(f"WARN: Airtable write failed ({e}); enqueuing offline.") self._enqueue_pending(fields_payload) return None def _enqueue_pending(self, payload: dict) -> None: self._pending_path.parent.mkdir(parents=True, exist_ok=True) existing: list[dict] = [] if self._pending_path.exists(): existing = json.loads(self._pending_path.read_text(encoding="utf-8")) existing.append({"queued_at": _now_iso(), "fields": payload}) self._pending_path.write_text(json.dumps(existing, indent=2), encoding="utf-8") def _emit_audit_event( self, event_type: str, linked_table: str, linked_id: str, confidence: float | None = None, notes: str = "", ) -> None: if self.offline: return try: f = FIELD["_Internal_Audit_Trail"] payload: dict[str, Any] = { f["id"]: str(uuid.uuid4()), f["event_type"]: event_type, f["linked_record_table"]: linked_table, f["linked_record_id"]: linked_id, f["agent_id"]: self.agent_id, f["agent_model"]: os.environ.get("LFCS_AGENT_MODEL", "claude-opus-4-7"), f["created_at"]: _now_iso(), f["notes"]: notes, } if confidence is not None: payload[f["confidence"]] = float(confidence) if self.session_record_id: payload[f["prompt_ref"]] = [self.session_record_id] at.create_records(BASE_ID, TBL["_Internal_Audit_Trail"], [payload]) except Exception as e: # noqa: BLE001 self._log(f"WARN: audit-trail emit failed ({e}).") def _mirror_to_vault(self, capture_record: dict) -> None: # Cross-job source of truth cross_job = ( VAULT_INTERNAL_ROOT / "audit-trail" / f"session-{self.session_ref}.md" ) cross_job.parent.mkdir(parents=True, exist_ok=True) with cross_job.open("a", encoding="utf-8") as fh: fh.write(self._format_capture_md(capture_record)) # Per-job filtered view if self.per_job_audit_dir: per_job = ( self.per_job_audit_dir / f"overrides-log-{dt.date.today().isoformat()}.md" ) per_job.parent.mkdir(parents=True, exist_ok=True) with per_job.open("a", encoding="utf-8") as fh: fh.write(self._format_capture_md(capture_record)) def _format_capture_md(self, capture_record: dict) -> str: f = FIELD["Bid_Line_Item_Overrides"] fields = capture_record["fields"] return ( f"\n## Override {capture_record.get('id', capture_record['uuid'])} — {_now_iso()}\n" f"- session_ref: `{self.session_ref}`\n" f"- bid_line_item: {fields[f['bid_line_item']]}\n" f"- original_rate_ex_gst: {fields[f['original_rate_ex_gst']]}\n" f"- override_rate_ex_gst: {fields[f['override_rate_ex_gst']]}\n" f"- reason_codes: {fields[f['reason_codes']]}\n" f"- reviewer_type: {fields[f['reviewer_type']]}\n" f"- reviewer_name: {fields[f['reviewer_name']]}\n" f"- free_text_rationale: {fields[f['free_text_rationale']]}\n" ) def _summary(self) -> str: n = len(self.captures) if n == 0: return "No overrides captured this session." net = sum( c["fields"][FIELD["Bid_Line_Item_Overrides"]["override_rate_ex_gst"]] - c["fields"][FIELD["Bid_Line_Item_Overrides"]["original_rate_ex_gst"]] for c in self.captures ) return f"{n} overrides captured. Net delta_dollars (sum of rate deltas) = {net:.2f}." def _log(self, msg: str) -> None: line = f"[{_now_iso()}] {msg}\n" log_path = VAULT_INTERNAL_ROOT / "prompt-logs" / f"session-{self.session_ref}.log" log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open("a", encoding="utf-8") as fh: fh.write(line)