1
from datetime import datetime
2
+from datetime import timedelta
3
+from difflib import SequenceMatcher
4
from typing import List
5
from typing import Optional
6
7
from fastapi import HTTPException
8
from loguru import logger
9
+from sqlalchemy import case
10
+from sqlalchemy import func
11
from sqlalchemy.ext.asyncio import AsyncSession
12
from sqlalchemy.future import select
13
+from sqlalchemy.orm import selectinload
14
15
from app.ai_analyst.schema.ai_analyst import AlertWithReportResponse
16
from app.ai_analyst.schema.ai_analyst import CreateJobRequest
17
from app.ai_analyst.schema.ai_analyst import CreateJobResponse
18
from app.ai_analyst.schema.ai_analyst import IocResponse
19
+from app.ai_analyst.schema.ai_analyst import IocReviewResponse
20
from app.ai_analyst.schema.ai_analyst import JobResponse
21
+from app.ai_analyst.schema.ai_analyst import MyReviewResponse
22
+from app.ai_analyst.schema.ai_analyst import PalaceConsolidationDuplicatePair
23
+from app.ai_analyst.schema.ai_analyst import PalaceConsolidationLesson
24
+from app.ai_analyst.schema.ai_analyst import PalaceConsolidationResponse
25
+from app.ai_analyst.schema.ai_analyst import PalaceConsolidationRoomGroup
26
+from app.ai_analyst.schema.ai_analyst import PalaceLessonResponse
27
+from app.ai_analyst.schema.ai_analyst import QueuePalaceLessonRequest
28
+from app.ai_analyst.schema.ai_analyst import QueuePalaceLessonResponse
29
from app.ai_analyst.schema.ai_analyst import ReportResponse
30
+from app.ai_analyst.schema.ai_analyst import ReviewResponse
31
+from app.ai_analyst.schema.ai_analyst import ReviewStatsIocAccuracy
32
+from app.ai_analyst.schema.ai_analyst import ReviewStatsResponse
33
+from app.ai_analyst.schema.ai_analyst import ReviewStatsTemplate
34
from app.ai_analyst.schema.ai_analyst import SubmitIocsRequest
35
from app.ai_analyst.schema.ai_analyst import SubmitIocsResponse
36
from app.ai_analyst.schema.ai_analyst import SubmitReportRequest
37
from app.ai_analyst.schema.ai_analyst import SubmitReportResponse
38
+from app.ai_analyst.schema.ai_analyst import SubmitReviewRequest
39
+from app.ai_analyst.schema.ai_analyst import SubmitReviewResponse
40
from app.ai_analyst.schema.ai_analyst import UpdateJobRequest
41
from app.ai_analyst.schema.ai_analyst import UpdateJobResponse
42
from app.db.universal_models import AiAnalystIoc
43
+from app.db.universal_models import AiAnalystIocReview
44
from app.db.universal_models import AiAnalystJob
45
+from app.db.universal_models import AiAnalystPalaceLesson
46
from app.db.universal_models import AiAnalystReport
47
+from app.db.universal_models import AiAnalystReview
48
from app.incidents.models import Alert
49
50
405
_report_to_response(report) if report else None,
406
[_ioc_to_response(i) for i in iocs],
407
)
408
+
409
+
410
+# --- Review / Palace lesson helpers ---
411
+
412
+
413
+def _ioc_review_to_response(ir: AiAnalystIocReview) -> IocReviewResponse:
414
+ return IocReviewResponse(
415
+ id=ir.id,
416
+ review_id=ir.review_id,
417
+ ioc_id=ir.ioc_id,
418
+ verdict_correct=ir.verdict_correct,
419
+ note=ir.note,
420
+ created_at=ir.created_at,
421
+ )
422
+
423
+
424
+def _review_to_response(review: AiAnalystReview) -> ReviewResponse:
425
+ return ReviewResponse(
426
+ id=review.id,
427
+ report_id=review.report_id,
428
+ alert_id=review.alert_id,
429
+ customer_code=review.customer_code,
430
+ reviewer_user_id=review.reviewer_user_id,
431
+ overall_verdict=review.overall_verdict,
432
+ template_choice=review.template_choice,
433
+ template_used=review.template_used,
434
+ rating_instructions=review.rating_instructions,
435
+ rating_artifacts=review.rating_artifacts,
436
+ rating_severity=review.rating_severity,
437
+ missing_steps=review.missing_steps,
438
+ suggested_edits=review.suggested_edits,
439
+ created_at=review.created_at,
440
+ updated_at=review.updated_at,
441
+ ioc_reviews=[_ioc_review_to_response(ir) for ir in (review.ioc_reviews or [])],
442
+ )
443
+
444
+
445
+def _palace_lesson_to_response(lesson: AiAnalystPalaceLesson) -> PalaceLessonResponse:
446
+ return PalaceLessonResponse(
447
+ id=lesson.id,
448
+ review_id=lesson.review_id,
449
+ customer_code=lesson.customer_code,
450
+ lesson_type=lesson.lesson_type,
451
+ lesson_text=lesson.lesson_text,
452
+ durability=lesson.durability,
453
+ status=lesson.status,
454
+ ingested_at=lesson.ingested_at,
455
+ created_at=lesson.created_at,
456
+ )
457
+
458
+
459
+async def submit_review(
460
+ report_id: int,
461
+ request: SubmitReviewRequest,
462
+ reviewer_user_id: int,
463
+ session: AsyncSession,
464
+) -> SubmitReviewResponse:
465
+ """
466
+ Upsert an analyst review of an AI investigation report.
467
+
468
+ Enforces one review per (report_id, reviewer_user_id) pair. If the user
469
+ has already reviewed this report, their existing row is updated in place
470
+ (and `updated_at` is set) and their per-IOC corrections are replaced
471
+ wholesale. Otherwise a new row is inserted.
472
+ """
473
+ logger.info(f"Submitting review for report {report_id} by user {reviewer_user_id}")
474
+
475
+ report = await session.get(AiAnalystReport, report_id)
476
+ if not report:
477
+ raise HTTPException(status_code=404, detail=f"Report {report_id} not found")
478
+
479
+ # If template_used wasn't supplied, inherit from the job for auditability
480
+ template_used = request.template_used
481
+ if template_used is None:
482
+ job = await session.get(AiAnalystJob, report.job_id)
483
+ if job is not None:
484
+ template_used = job.template_used
485
+
486
+ # Validate every referenced IOC before mutating anything. A 400 here means
487
+ # we don't half-write and then bail.
488
+ for correction in request.ioc_reviews:
489
+ ioc = await session.get(AiAnalystIoc, correction.ioc_id)
490
+ if ioc is None or ioc.report_id != report_id:
491
+ raise HTTPException(
492
+ status_code=400,
493
+ detail=f"IOC {correction.ioc_id} not found or does not belong to report {report_id}",
494
+ )
495
+
496
+ # Look up existing review by the unique (report_id, reviewer_user_id) key
497
+ existing_result = await session.execute(
498
+ select(AiAnalystReview)
499
+ .where(AiAnalystReview.report_id == report_id)
500
+ .where(AiAnalystReview.reviewer_user_id == reviewer_user_id)
501
+ .options(selectinload(AiAnalystReview.ioc_reviews)),
502
+ )
503
+ review = existing_result.scalars().first()
504
+
505
+ is_edit = review is not None
506
+
507
+ if is_edit:
508
+ # Update existing review in place
509
+ review.overall_verdict = request.overall_verdict.value if request.overall_verdict else None
510
+ review.template_choice = request.template_choice.value if request.template_choice else None
511
+ review.template_used = template_used
512
+ review.rating_instructions = request.rating_instructions
513
+ review.rating_artifacts = request.rating_artifacts
514
+ review.rating_severity = request.rating_severity
515
+ review.missing_steps = request.missing_steps
516
+ review.suggested_edits = request.suggested_edits
517
+ review.updated_at = datetime.utcnow()
518
+ session.add(review)
519
+
520
+ # Replace per-IOC corrections wholesale — simpler than diffing and the
521
+ # edit UI always submits the full set anyway.
522
+ for old_ir in list(review.ioc_reviews or []):
523
+ await session.delete(old_ir)
524
+ await session.flush()
525
+ else:
526
+ review = AiAnalystReview(
527
+ report_id=report_id,
528
+ alert_id=report.alert_id,
529
+ customer_code=report.customer_code,
530
+ reviewer_user_id=reviewer_user_id,
531
+ overall_verdict=request.overall_verdict.value if request.overall_verdict else None,
532
+ template_choice=request.template_choice.value if request.template_choice else None,
533
+ template_used=template_used,
534
+ rating_instructions=request.rating_instructions,
535
+ rating_artifacts=request.rating_artifacts,
536
+ rating_severity=request.rating_severity,
537
+ missing_steps=request.missing_steps,
538
+ suggested_edits=request.suggested_edits,
539
+ created_at=datetime.utcnow(),
540
+ )
541
+ session.add(review)
542
+ await session.flush() # get review.id before inserting child rows
543
+
544
+ # Insert the new per-IOC corrections
545
+ for correction in request.ioc_reviews:
546
+ session.add(
547
+ AiAnalystIocReview(
548
+ review_id=review.id,
549
+ ioc_id=correction.ioc_id,
550
+ verdict_correct=correction.verdict_correct,
551
+ note=correction.note,
552
+ created_at=datetime.utcnow(),
553
+ ),
554
+ )
555
+
556
+ await session.commit()
557
+
558
+ # Re-fetch with ioc_reviews eagerly loaded for the response
559
+ result = await session.execute(
560
+ select(AiAnalystReview).where(AiAnalystReview.id == review.id).options(selectinload(AiAnalystReview.ioc_reviews)),
561
+ )
562
+ review_loaded = result.scalars().first()
563
+
564
+ action = "updated" if is_edit else "created"
565
+ logger.info(f"Review {review.id} {action} for report {report_id}")
566
+ return SubmitReviewResponse(
567
+ success=True,
568
+ message=f"Review {action}",
569
+ review=_review_to_response(review_loaded),
570
+ )
571
+
572
+
573
+async def get_my_review(
574
+ report_id: int,
575
+ reviewer_user_id: int,
576
+ session: AsyncSession,
577
+) -> MyReviewResponse:
578
+ """
579
+ Look up the current user's existing review for a report.
580
+
581
+ Used by the UI to decide whether to render the rubric in 'create' mode or
582
+ 'edit existing' mode. Returns success=True with review=None when no review
583
+ exists yet (not an error).
584
+ """
585
+ # Ensure the report itself exists so the UI gets a real 404 for bad ids
586
+ report = await session.get(AiAnalystReport, report_id)
587
+ if not report:
588
+ raise HTTPException(status_code=404, detail=f"Report {report_id} not found")
589
+
590
+ result = await session.execute(
591
+ select(AiAnalystReview)
592
+ .where(AiAnalystReview.report_id == report_id)
593
+ .where(AiAnalystReview.reviewer_user_id == reviewer_user_id)
594
+ .options(selectinload(AiAnalystReview.ioc_reviews)),
595
+ )
596
+ review = result.scalars().first()
597
+
598
+ if review is None:
599
+ return MyReviewResponse(
600
+ success=True,
601
+ message="No existing review for this user on this report",
602
+ review=None,
603
+ )
604
+
605
+ return MyReviewResponse(
606
+ success=True,
607
+ message="Existing review retrieved",
608
+ review=_review_to_response(review),
609
+ )
610
+
611
+
612
+async def queue_palace_lesson(
613
+ request: QueuePalaceLessonRequest,
614
+ session: AsyncSession,
615
+) -> QueuePalaceLessonResponse:
616
+ """
617
+ Queue a MemPalace lesson for async drainer pickup. Does NOT call Talon
618
+ directly — the drainer (roadmap item 17) reads status='pending' rows and
619
+ POSTs to NanoClaw's /palace/lesson endpoint.
620
+ """
621
+ logger.info(f"Queuing palace lesson for {request.customer_code})")
622
+
623
+ # If review_id supplied, validate it exists
624
+ if request.review_id is not None:
625
+ review = await session.get(AiAnalystReview, request.review_id)
626
+ if review is None:
627
+ raise HTTPException(
628
+ status_code=404,
629
+ detail=f"Review {request.review_id} not found",
630
+ )
631
+
632
+ lesson = AiAnalystPalaceLesson(
633
+ review_id=request.review_id,
634
+ customer_code=request.customer_code,
635
+ lesson_type=request.lesson_type.value,
636
+ lesson_text=request.lesson_text,
637
+ durability=request.durability.value,
638
+ status="pending",
639
+ created_at=datetime.utcnow(),
640
+ )
641
+ session.add(lesson)
642
+ await session.commit()
643
+ await session.refresh(lesson)
644
+
645
+ logger.info(f"Palace lesson {lesson.id} queued (status=pending)")
646
+ return QueuePalaceLessonResponse(
647
+ success=True,
648
+ message="Palace lesson queued for ingestion",
649
+ lesson=_palace_lesson_to_response(lesson),
650
+ )
651
+
652
+
653
+async def list_reviews_by_customer(
654
+ customer_code: str,
655
+ session: AsyncSession,
656
+) -> List[ReviewResponse]:
657
+ """Dashboard feed — reviews for a customer, newest first, with nested IOC reviews."""
658
+ result = await session.execute(
659
+ select(AiAnalystReview)
660
+ .where(AiAnalystReview.customer_code == customer_code)
661
+ .options(selectinload(AiAnalystReview.ioc_reviews))
662
+ .order_by(AiAnalystReview.created_at.desc()),
663
+ )
664
+ reviews = result.scalars().all()
665
+ return [_review_to_response(r) for r in reviews]
666
+
667
+
668
+def _pct(numerator: int, denominator: int) -> Optional[float]:
669
+ """Percentage helper — None when the denominator is 0 so the UI can
670
+ render a dash instead of a misleading '0%'."""
671
+ if denominator <= 0:
672
+ return None
673
+ return round((numerator / denominator) * 100, 2)
674
+
675
+
676
+def _round_float(v) -> Optional[float]:
677
+ """Avg helper — SQLAlchemy returns Decimal on some dialects; normalize to
678
+ a 2-decimal float for JSON. Returns None if the source was NULL (no rows)."""
679
+ if v is None:
680
+ return None
681
+ return round(float(v), 2)
682
+
683
+
684
+async def get_review_stats(
685
+ customer_code: str,
686
+ session: AsyncSession,
687
+ recent_limit: int = 10,
688
+) -> ReviewStatsResponse:
689
+ """Aggregate feedback metrics for the customer's review dashboard.
690
+
691
+ Uses SQL-side COUNT/AVG/CASE aggregates so this scales with review count
692
+ rather than pulling every row through Python — per the scale-first
693
+ design call on Step 20.
694
+ """
695
+ # Main rollup: totals, verdict counts, template-choice counts, avg ratings.
696
+ main_q = select(
697
+ func.count(AiAnalystReview.id).label("total"),
698
+ func.sum(case((AiAnalystReview.overall_verdict == "up", 1), else_=0)).label("thumbs_up"),
699
+ func.sum(case((AiAnalystReview.overall_verdict == "down", 1), else_=0)).label("thumbs_down"),
700
+ func.sum(case((AiAnalystReview.template_choice == "correct", 1), else_=0)).label("tpl_correct"),
701
+ func.sum(case((AiAnalystReview.template_choice == "partial", 1), else_=0)).label("tpl_partial"),
702
+ func.sum(case((AiAnalystReview.template_choice == "wrong", 1), else_=0)).label("tpl_wrong"),
703
+ func.avg(AiAnalystReview.rating_instructions).label("avg_instr"),
704
+ func.avg(AiAnalystReview.rating_artifacts).label("avg_artifacts"),
705
+ func.avg(AiAnalystReview.rating_severity).label("avg_severity"),
706
+ ).where(AiAnalystReview.customer_code == customer_code)
707
+ main_row = (await session.execute(main_q)).one()
708
+
709
+ total = int(main_row.total or 0)
710
+ thumbs_up = int(main_row.thumbs_up or 0)
711
+ thumbs_down = int(main_row.thumbs_down or 0)
712
+ # Non-null denominator for the up% gauge — reviews that actually picked a
713
+ # thumb. Skips reviews that left overall_verdict null.
714
+ verdict_total = thumbs_up + thumbs_down
715
+
716
+ # Per-template rollup, grouped by template_used (which may be NULL).
717
+ per_template_q = (
718
+ select(
719
+ AiAnalystReview.template_used.label("template_used"),
720
+ func.count(AiAnalystReview.id).label("total"),
721
+ func.sum(case((AiAnalystReview.overall_verdict == "up", 1), else_=0)).label("thumbs_up"),
722
+ func.sum(case((AiAnalystReview.overall_verdict == "down", 1), else_=0)).label("thumbs_down"),
723
+ func.sum(case((AiAnalystReview.template_choice == "correct", 1), else_=0)).label("correct"),
724
+ func.sum(case((AiAnalystReview.template_choice == "partial", 1), else_=0)).label("partial"),
725
+ func.sum(case((AiAnalystReview.template_choice == "wrong", 1), else_=0)).label("wrong"),
726
+ func.avg(AiAnalystReview.rating_instructions).label("avg_instr"),
727
+ func.avg(AiAnalystReview.rating_artifacts).label("avg_artifacts"),
728
+ func.avg(AiAnalystReview.rating_severity).label("avg_severity"),
729
+ )
730
+ .where(AiAnalystReview.customer_code == customer_code)
731
+ .group_by(AiAnalystReview.template_used)
732
+ .order_by(func.count(AiAnalystReview.id).desc())
733
+ )
734
+ per_template_rows = (await session.execute(per_template_q)).all()
735
+
736
+ # IOC verdict accuracy — join IocReview rows back to the customer's reviews
737
+ # so we only count corrections attached to this customer's reports.
738
+ ioc_q = (
739
+ select(
740
+ func.count(AiAnalystIocReview.id).label("total"),
741
+ func.sum(case((AiAnalystIocReview.verdict_correct.is_(True), 1), else_=0)).label("correct"),
742
+ func.sum(case((AiAnalystIocReview.verdict_correct.is_(False), 1), else_=0)).label("incorrect"),
743
+ )
744
+ .join(AiAnalystReview, AiAnalystReview.id == AiAnalystIocReview.review_id)
745
+ .where(AiAnalystReview.customer_code == customer_code)
746
+ )
747
+ ioc_row = (await session.execute(ioc_q)).one()
748
+ ioc_total = int(ioc_row.total or 0)
749
+ ioc_correct = int(ioc_row.correct or 0)
750
+ ioc_incorrect = int(ioc_row.incorrect or 0)
751
+
752
+ # Recent reviews (hydrated with ioc_reviews for drill-in).
753
+ recent_q = (
754
+ select(AiAnalystReview)
755
+ .where(AiAnalystReview.customer_code == customer_code)
756
+ .options(selectinload(AiAnalystReview.ioc_reviews))
757
+ .order_by(AiAnalystReview.created_at.desc())
758
+ .limit(recent_limit)
759
+ )
760
+ recent_reviews = (await session.execute(recent_q)).scalars().all()
761
+
762
+ per_template: List[ReviewStatsTemplate] = [
763
+ ReviewStatsTemplate(
764
+ template_used=row.template_used,
765
+ total=int(row.total or 0),
766
+ thumbs_up=int(row.thumbs_up or 0),
767
+ thumbs_down=int(row.thumbs_down or 0),
768
+ correct=int(row.correct or 0),
769
+ partial=int(row.partial or 0),
770
+ wrong=int(row.wrong or 0),
771
+ avg_rating_instructions=_round_float(row.avg_instr),
772
+ avg_rating_artifacts=_round_float(row.avg_artifacts),
773
+ avg_rating_severity=_round_float(row.avg_severity),
774
+ )
775
+ for row in per_template_rows
776
+ ]
777
+
778
+ return ReviewStatsResponse(
779
+ success=True,
780
+ message=f"Review stats for {customer_code}",
781
+ customer_code=customer_code,
782
+ total_reviews=total,
783
+ thumbs_up=thumbs_up,
784
+ thumbs_down=thumbs_down,
785
+ thumbs_up_pct=_pct(thumbs_up, verdict_total),
786
+ template_choice_correct=int(main_row.tpl_correct or 0),
787
+ template_choice_partial=int(main_row.tpl_partial or 0),
788
+ template_choice_wrong=int(main_row.tpl_wrong or 0),
789
+ avg_rating_instructions=_round_float(main_row.avg_instr),
790
+ avg_rating_artifacts=_round_float(main_row.avg_artifacts),
791
+ avg_rating_severity=_round_float(main_row.avg_severity),
792
+ ioc_accuracy=ReviewStatsIocAccuracy(
793
+ total=ioc_total,
794
+ correct=ioc_correct,
795
+ incorrect=ioc_incorrect,
796
+ accuracy_pct=_pct(ioc_correct, ioc_total),
797
+ ),
798
+ per_template=per_template,
799
+ recent_reviews=[_review_to_response(r) for r in recent_reviews],
800
+ )
801
+
802
+
803
+# --- Palace consolidation (Step 21.B) ---
804
+
805
+# Keep in sync with invoke_palace_lesson_sweeper.ONE_OFF_EXPIRY_DAYS —
806
+# duplicated here rather than imported to avoid pulling a scheduler
807
+# service into the synchronous route path at import time.
808
+_ONE_OFF_EXPIRY_DAYS = 7
809
+# Lessons within this many days of expiring are surfaced to the reviewer
810
+# as "about to be swept" — gives a window to promote to durable.
811
+_EXPIRY_SOON_WINDOW_DAYS = 2
812
+# Similarity threshold for near-duplicate detection. 0.70 picks up
813
+# paraphrases without flooding the reviewer with every shared phrase.
814
+# difflib's SequenceMatcher on short strings is cheap — we can afford
815
+# O(n²) pairs per room.
816
+_DUPLICATE_SIMILARITY_THRESHOLD = 0.70
817
+
818
+
819
+def _lesson_to_consolidation(
820
+ lesson: AiAnalystPalaceLesson,
821
+ now: datetime,
822
+) -> PalaceConsolidationLesson:
823
+ days_until_expiry: Optional[int] = None
824
+ if lesson.durability == "one_off" and lesson.ingested_at is not None:
825
+ expiry = lesson.ingested_at + timedelta(days=_ONE_OFF_EXPIRY_DAYS)
826
+ days_until_expiry = (expiry - now).days
827
+ return PalaceConsolidationLesson(
828
+ id=lesson.id,
829
+ lesson_type=lesson.lesson_type,
830
+ lesson_text=lesson.lesson_text,
831
+ durability=lesson.durability,
832
+ status=lesson.status,
833
+ drawer_id=lesson.drawer_id,
834
+ created_at=lesson.created_at,
835
+ ingested_at=lesson.ingested_at,
836
+ days_until_expiry=days_until_expiry,
837
+ )
838
+
839
+
840
+def _find_duplicate_pairs(
841
+ lessons: List[PalaceConsolidationLesson],
842
+) -> List[PalaceConsolidationDuplicatePair]:
843
+ """Pairwise SequenceMatcher within the same room. Only returns
844
+ pairs above the threshold, sorted by similarity descending."""
845
+ pairs: List[PalaceConsolidationDuplicatePair] = []
846
+ # Group by room first so we only compare within-room.
847
+ by_room: dict[str, List[PalaceConsolidationLesson]] = {}
848
+ for lesson in lessons:
849
+ by_room.setdefault(lesson.lesson_type, []).append(lesson)
850
+
851
+ for room, room_lessons in by_room.items():
852
+ # Normalize once up-front so SequenceMatcher has stable inputs.
853
+ normalized = [(ls, ls.lesson_text.strip().lower()) for ls in room_lessons]
854
+ for i in range(len(normalized)):
855
+ a_lesson, a_text = normalized[i]
856
+ if not a_text:
857
+ continue
858
+ for j in range(i + 1, len(normalized)):
859
+ b_lesson, b_text = normalized[j]
860
+ if not b_text:
861
+ continue
862
+ ratio = SequenceMatcher(None, a_text, b_text).ratio()
863
+ if ratio >= _DUPLICATE_SIMILARITY_THRESHOLD:
864
+ pairs.append(
865
+ PalaceConsolidationDuplicatePair(
866
+ room=room,
867
+ lesson_a_id=a_lesson.id,
868
+ lesson_b_id=b_lesson.id,
869
+ lesson_a_text=a_lesson.lesson_text,
870
+ lesson_b_text=b_lesson.lesson_text,
871
+ similarity=round(ratio, 3),
872
+ ),
873
+ )
874
+ pairs.sort(key=lambda p: p.similarity, reverse=True)
875
+ return pairs
876
+
877
+
878
+def _render_consolidation_markdown(
879
+ customer_code: str,
880
+ generated_at: datetime,
881
+ total_lessons: int,
882
+ total_durable: int,
883
+ total_one_off: int,
884
+ rooms: List[PalaceConsolidationRoomGroup],
885
+ duplicates: List[PalaceConsolidationDuplicatePair],
886
+ upcoming: List[PalaceConsolidationLesson],
887
+) -> str:
888
+ """Pre-render the digest as markdown so the drawer can offer a
889
+ one-click copy/export. Kept intentionally terse — headings + bullets."""
890
+ lines: List[str] = []
891
+ lines.append(f"# Palace consolidation — {customer_code}")
892
+ lines.append(f"_Generated {generated_at.isoformat()} UTC_")
893
+ lines.append("")
894
+ lines.append("## Summary")
895
+ lines.append(f"- Total active lessons: **{total_lessons}**")
896
+ lines.append(f"- Durable: {total_durable} | One-off: {total_one_off}")
897
+ if upcoming:
898
+ lines.append(
899
+ f"- **{len(upcoming)} one-off lesson(s) expiring within "
900
+ f"{_EXPIRY_SOON_WINDOW_DAYS} day(s)** — consider promoting to durable.",
901
+ )
902
+ if duplicates:
903
+ lines.append(f"- **{len(duplicates)} near-duplicate pair(s)** flagged for review.")
904
+ lines.append("")
905
+
906
+ if upcoming:
907
+ lines.append("## Upcoming expirations")
908
+ for ls in upcoming:
909
+ due = ls.days_until_expiry if ls.days_until_expiry is not None else "?"
910
+ lines.append(f"- _{ls.lesson_type}_ (id {ls.id}, in {due}d): {ls.lesson_text}")
911
+ lines.append("")
912
+
913
+ if duplicates:
914
+ lines.append("## Near-duplicate candidates")
915
+ for pair in duplicates:
916
+ pct = int(pair.similarity * 100)
917
+ lines.append(f"- **{pair.room}** — {pct}% similar")
918
+ lines.append(f" - #{pair.lesson_a_id}: {pair.lesson_a_text}")
919
+ lines.append(f" - #{pair.lesson_b_id}: {pair.lesson_b_text}")
920
+ lines.append("")
921
+
922
+ lines.append("## Rooms")
923
+ for group in rooms:
924
+ lines.append(
925
+ f"### {group.room} ({group.total} total — " f"{group.durable} durable, {group.one_off} one-off)",
926
+ )
927
+ for ls in group.lessons:
928
+ tag = "🧷" if ls.durability == "durable" else "⏳"
929
+ suffix = ""
930
+ if ls.durability == "one_off" and ls.days_until_expiry is not None:
931
+ suffix = f" _(expires in {ls.days_until_expiry}d)_"
932
+ lines.append(f"- {tag} #{ls.id}{suffix}: {ls.lesson_text}")
933
+ lines.append("")
934
+
935
+ return "\n".join(lines).rstrip() + "\n"
936
+
937
+
938
+async def get_palace_consolidation(
939
+ customer_code: str,
940
+ session: AsyncSession,
941
+) -> PalaceConsolidationResponse:
942
+ """Build a point-in-time digest of a customer's active MemPalace
943
+ lessons. Pure read-only, pure Python — no Talon round-trip. Used by
944
+ the manual "Consolidate Lessons" button in the Feedback dashboard."""
945
+ logger.info(f"Building palace consolidation digest for customer {customer_code}")
946
+
947
+ # Exclude expired (already swept) and failed (never reached the palace)
948
+ # rows — consolidation is about what's actually live right now.
949
+ stmt = (
950
+ select(AiAnalystPalaceLesson)
951
+ .where(AiAnalystPalaceLesson.customer_code == customer_code)
952
+ .where(AiAnalystPalaceLesson.status.in_(["pending", "ingested"]))
953
+ .order_by(
954
+ AiAnalystPalaceLesson.lesson_type.asc(),
955
+ AiAnalystPalaceLesson.created_at.desc(),
956
+ )
957
+ )
958
+ result = await session.execute(stmt)
959
+ raw_lessons = result.scalars().all()
960
+
961
+ now = datetime.utcnow()
962
+ lessons = [_lesson_to_consolidation(ls, now) for ls in raw_lessons]
963
+
964
+ total_lessons = len(lessons)
965
+ total_durable = sum(1 for ls in lessons if ls.durability == "durable")
966
+ total_one_off = sum(1 for ls in lessons if ls.durability == "one_off")
967
+ total_pending = sum(1 for ls in lessons if ls.status == "pending")
968
+ total_ingested = sum(1 for ls in lessons if ls.status == "ingested")
969
+
970
+ upcoming = sorted(
971
+ [
972
+ ls
973
+ for ls in lessons
974
+ if ls.durability == "one_off" and ls.days_until_expiry is not None and ls.days_until_expiry <= _EXPIRY_SOON_WINDOW_DAYS
975
+ ],
976
+ key=lambda ls: ls.days_until_expiry if ls.days_until_expiry is not None else 0,
977
+ )
978
+
979
+ # Build per-room groups in deterministic room order.
980
+ by_room: dict[str, List[PalaceConsolidationLesson]] = {}
981
+ for ls in lessons:
982
+ by_room.setdefault(ls.lesson_type, []).append(ls)
983
+ rooms: List[PalaceConsolidationRoomGroup] = []
984
+ for room in sorted(by_room.keys()):
985
+ room_lessons = by_room[room]
986
+ rooms.append(
987
+ PalaceConsolidationRoomGroup(
988
+ room=room,
989
+ total=len(room_lessons),
990
+ durable=sum(1 for ls in room_lessons if ls.durability == "durable"),
991
+ one_off=sum(1 for ls in room_lessons if ls.durability == "one_off"),
992
+ lessons=room_lessons,
993
+ ),
994
+ )
995
+
996
+ duplicates = _find_duplicate_pairs(lessons)
997
+
998
+ markdown = _render_consolidation_markdown(
999
+ customer_code=customer_code,
1000
+ generated_at=now,
1001
+ total_lessons=total_lessons,
1002
+ total_durable=total_durable,
1003
+ total_one_off=total_one_off,
1004
+ rooms=rooms,
1005
+ duplicates=duplicates,
1006
+ upcoming=upcoming,
1007
+ )
1008
+
1009
+ return PalaceConsolidationResponse(
1010
+ success=True,
1011
+ message=(
1012
+ f"Palace consolidation for {customer_code}: "
1013
+ f"{total_lessons} active lesson(s), "
1014
+ f"{len(duplicates)} duplicate pair(s), "
1015
+ f"{len(upcoming)} expiring soon"
1016
+ ),
1017
+ customer_code=customer_code,
1018
+ generated_at=now,
1019
+ total_lessons=total_lessons,
1020
+ total_durable=total_durable,
1021
+ total_one_off=total_one_off,
1022
+ total_pending=total_pending,
1023
+ total_ingested=total_ingested,
1024
+ upcoming_expirations=upcoming,
1025
+ rooms=rooms,
1026
+ duplicate_candidates=duplicates,
1027
+ markdown=markdown,
1028
+ )