| 1 | """ |
| 2 | Palace lesson drainer — step 17 of the CoPilot ↔ NanoClaw Talon integration. |
| 3 | |
| 4 | Polls the `ai_analyst_palace_lesson` table for rows with status='pending' and |
| 5 | POSTs each one to NanoClaw's `/palace/lesson` endpoint (which wraps the |
| 6 | MemPalace `add_drawer` MCP tool). On success the row is marked 'ingested' |
| 7 | with an `ingested_at` timestamp; on any failure the row is marked 'failed' |
| 8 | and left alone — no automatic retry. The teach-the-palace UI surfaces |
| 9 | failures so an operator can manually requeue. |
| 10 | |
| 11 | Scheduling: runs every 2 minutes via APScheduler. Batch size is capped to |
| 12 | prevent a large backlog from locking the scheduler tick. |
| 13 | """ |
| 14 | from datetime import datetime |
| 15 | |
| 16 | from loguru import logger |
| 17 | from sqlalchemy.future import select |
| 18 | |
| 19 | from app.connectors.talon.utils.universal import send_post_request |
| 20 | from app.db.db_session import get_db_session |
| 21 | from app.db.universal_models import AiAnalystPalaceLesson |
| 22 | from app.schedulers.models.scheduler import JobMetadata |
| 23 | |
| 24 | JOB_ID = "invoke_palace_lesson_drainer" |
| 25 | |
| 26 | # How many pending lessons to process per scheduler tick. A large backlog |
| 27 | # gets drained over multiple ticks rather than hogging the event loop. |
| 28 | DEFAULT_BATCH_SIZE = 25 |
| 29 | |
| 30 | |
| 31 | async def invoke_palace_lesson_drainer() -> None: |
| 32 | """ |
| 33 | Drain one batch of pending palace lessons to NanoClaw. |
| 34 | |
| 35 | Returns silently (logging only) in these cases: |
| 36 | - No pending lessons exist. |
| 37 | - Talon connector is not configured in the DB. |
| 38 | - The HTTP POST raises or returns success=False (row → 'failed'). |
| 39 | |
| 40 | Exceptions here must not propagate up into APScheduler — the EVENT_JOB_ERROR |
| 41 | listener in scheduler.py would log a crash, and the job would continue on |
| 42 | its interval anyway. |
| 43 | """ |
| 44 | logger.info("Palace lesson drainer tick") |
| 45 | |
| 46 | async with get_db_session() as session: |
| 47 | # Pull oldest pending lessons first; cap the batch |
| 48 | stmt = ( |
| 49 | select(AiAnalystPalaceLesson) |
| 50 | .where(AiAnalystPalaceLesson.status == "pending") |
| 51 | .order_by(AiAnalystPalaceLesson.created_at.asc()) |
| 52 | .limit(DEFAULT_BATCH_SIZE) |
| 53 | ) |
| 54 | result = await session.execute(stmt) |
| 55 | lessons = result.scalars().all() |
| 56 | |
| 57 | if not lessons: |
| 58 | logger.debug("No pending palace lessons to drain") |
| 59 | await _mark_job_success(session) |
| 60 | return |
| 61 | |
| 62 | logger.info(f"Draining {len(lessons)} pending palace lesson(s) to NanoClaw") |
| 63 | |
| 64 | ingested = 0 |
| 65 | failed = 0 |
| 66 | |
| 67 | for lesson in lessons: |
| 68 | payload = { |
| 69 | "customer_code": lesson.customer_code, |
| 70 | "lesson_type": lesson.lesson_type, |
| 71 | "lesson_text": lesson.lesson_text, |
| 72 | "durability": lesson.durability, |
| 73 | } |
| 74 | |
| 75 | try: |
| 76 | # send_post_request handles connector lookup + auth headers + |
| 77 | # error trapping. It never raises; it returns {success, ...}. |
| 78 | response = await send_post_request( |
| 79 | endpoint="/palace/lesson", |
| 80 | data=payload, |
| 81 | timeout=60, |
| 82 | ) |
| 83 | except Exception as e: |
| 84 | # Defensive — shouldn't happen because send_post_request |
| 85 | # already wraps its own exceptions, but we don't want one |
| 86 | # bad lesson to break the whole batch. |
| 87 | logger.error(f"Unexpected error posting palace lesson {lesson.id}: {e}") |
| 88 | response = {"success": False, "message": str(e)} |
| 89 | |
| 90 | if response.get("success"): |
| 91 | lesson.status = "ingested" |
| 92 | lesson.ingested_at = datetime.utcnow() |
| 93 | # Capture drawer_id so the durability sweeper can later |
| 94 | # call /palace/forget for expired one-off lessons. |
| 95 | # send_post_request wraps the raw NanoClaw body under |
| 96 | # response["data"]; mempalace's tool_add_drawer places |
| 97 | # drawer_id at the top of its return dict. |
| 98 | body = response.get("data") if isinstance(response.get("data"), dict) else {} |
| 99 | drawer_id = body.get("drawer_id") |
| 100 | if isinstance(drawer_id, str) and drawer_id: |
| 101 | lesson.drawer_id = drawer_id |
| 102 | else: |
| 103 | logger.warning( |
| 104 | f"Palace lesson {lesson.id} ingested without drawer_id in response " |
| 105 | f"(body_keys={list(body.keys()) if body else []}); sweeper will skip this row", |
| 106 | ) |
| 107 | ingested += 1 |
| 108 | logger.info(f"Lesson {lesson.id} ingested (customer={lesson.customer_code}, type={lesson.lesson_type})") |
| 109 | else: |
| 110 | lesson.status = "failed" |
| 111 | failed += 1 |
| 112 | logger.warning( |
| 113 | f"Palace lesson {lesson.id} failed: " f"{response.get('message', 'unknown error')}", |
| 114 | ) |
| 115 | |
| 116 | session.add(lesson) |
| 117 | # Commit per-row so partial progress survives a crash mid-batch. |
| 118 | await session.commit() |
| 119 | |
| 120 | logger.info( |
| 121 | f"Palace lesson drainer complete: ingested={ingested}, failed={failed}", |
| 122 | ) |
| 123 | |
| 124 | await _mark_job_success(session) |
| 125 | |
| 126 | |
| 127 | async def _mark_job_success(session) -> None: |
| 128 | """Update JobMetadata.last_success for the drainer job.""" |
| 129 | stmt = select(JobMetadata).where(JobMetadata.job_id == JOB_ID) |
| 130 | result = await session.execute(stmt) |
| 131 | job_metadata = result.scalars().first() |
| 132 | if job_metadata: |
| 133 | job_metadata.last_success = datetime.utcnow() |
| 134 | session.add(job_metadata) |
| 135 | await session.commit() |
| 136 | else: |
| 137 | logger.warning(f"JobMetadata for {JOB_ID!r} not found") |