@cryptotaxi247 / CoPilot
Code Commits Issues Pulls Sessions CI/CD
main
main
py 146 lines 5.76 KB
Raw
1 """
2 Palace lesson durability sweeper — Step 21.A of the CoPilot ↔ NanoClaw
3 Talon integration.
4
5 Scans the ``ai_analyst_palace_lesson`` table for rows that meet all of:
6 - durability == 'one_off'
7 - status == 'ingested'
8 - drawer_id is not null
9 - ingested_at is older than ONE_OFF_EXPIRY_DAYS
10
11 For each match, POSTs {"drawer_id": ...} to NanoClaw's ``/palace/forget``
12 endpoint (which wraps ``mempalace.tool_delete_drawer``). On any response
13 — success or failure — the row is flipped to status='expired' with
14 ``expired_at = now()`` so it never gets re-processed. A failed forget is
15 logged as a warning but not retried: the lesson is past its shelf life
16 either way, and a stuck row would jam the sweeper forever.
17
18 Scheduling: hourly batch drain, capped at DEFAULT_BATCH_SIZE per tick to
19 keep the scheduler responsive if a big expiry wave lands at once.
20 """
21 from datetime import datetime
22 from datetime import timedelta
23
24 from loguru import logger
25 from sqlalchemy.future import select
26
27 from app.connectors.talon.utils.universal import send_post_request
28 from app.db.db_session import get_db_session
29 from app.db.universal_models import AiAnalystPalaceLesson
30 from app.schedulers.models.scheduler import JobMetadata
31
32 JOB_ID = "invoke_palace_lesson_sweeper"
33
34 # One-off lessons expire this many days after their ingested_at timestamp.
35 # Durable lessons are never swept — they live in MemPalace indefinitely
36 # until a human removes them manually.
37 ONE_OFF_EXPIRY_DAYS = 7
38
39 # Max rows forgotten per tick. A large backlog drains over multiple ticks
40 # instead of blocking the event loop or overwhelming NanoClaw.
41 DEFAULT_BATCH_SIZE = 25
42
43
44 async def invoke_palace_lesson_sweeper() -> None:
45 """
46 Forget one batch of expired one-off palace lessons.
47
48 Silently returns (logging only) when:
49 - No rows have aged past the expiry window.
50 - The Talon connector is not configured in the DB.
51 - ``/palace/forget`` returns success=False (row → 'expired' anyway).
52
53 Exceptions never propagate — APScheduler will re-invoke on the next
54 tick regardless, and the EVENT_JOB_ERROR listener noise is not useful
55 here when the fix is "wait for the next run."
56 """
57 logger.info("Palace lesson sweeper tick")
58
59 cutoff = datetime.utcnow() - timedelta(days=ONE_OFF_EXPIRY_DAYS)
60
61 async with get_db_session() as session:
62 stmt = (
63 select(AiAnalystPalaceLesson)
64 .where(AiAnalystPalaceLesson.durability == "one_off")
65 .where(AiAnalystPalaceLesson.status == "ingested")
66 .where(AiAnalystPalaceLesson.drawer_id.is_not(None))
67 .where(AiAnalystPalaceLesson.ingested_at.is_not(None))
68 .where(AiAnalystPalaceLesson.ingested_at < cutoff)
69 .order_by(AiAnalystPalaceLesson.ingested_at.asc())
70 .limit(DEFAULT_BATCH_SIZE)
71 )
72 result = await session.execute(stmt)
73 lessons = result.scalars().all()
74
75 if not lessons:
76 logger.debug("No expired one-off palace lessons to sweep")
77 await _mark_job_success(session)
78 return
79
80 logger.info(f"Sweeping {len(lessons)} expired one-off lessons (cutoff={cutoff.isoformat()})")
81
82 forgotten = 0
83 forget_failed = 0
84
85 for lesson in lessons:
86 payload = {"drawer_id": lesson.drawer_id}
87
88 try:
89 response = await send_post_request(
90 endpoint="/palace/forget",
91 data=payload,
92 timeout=30,
93 )
94 except Exception as e:
95 # Defensive — send_post_request already traps its own
96 # exceptions, but we never want one bad row to block the
97 # batch. Record the row as expired regardless.
98 logger.error(
99 f"Unexpected error forgetting palace lesson {lesson.id}: {e}",
100 )
101 response = {"success": False, "message": str(e)}
102
103 # Mempalace returns {success, drawer_id, error?} under the
104 # top-level "data" key when send_post_request succeeds.
105 body = response.get("data") if isinstance(response.get("data"), dict) else {}
106 mem_success = bool(body.get("success"))
107
108 if response.get("success") and mem_success:
109 forgotten += 1
110 logger.info(f"Lesson {lesson.id} forgotten (drawer_id={lesson.drawer_id}, customer={lesson.customer_code})")
111 else:
112 forget_failed += 1
113 logger.warning(
114 f"Palace lesson {lesson.id} forget failed "
115 f"(drawer_id={lesson.drawer_id}); "
116 f"flipping to expired anyway. "
117 f"transport_error={response.get('message')}, "
118 f"mem_error={body.get('error')}",
119 )
120
121 # Flip to expired either way — the lesson is past shelf life,
122 # and a stuck row would clog the sweeper on every future tick.
123 lesson.status = "expired"
124 lesson.expired_at = datetime.utcnow()
125 session.add(lesson)
126 # Commit per-row so partial progress survives a mid-batch crash.
127 await session.commit()
128
129 logger.info(
130 f"Palace lesson sweeper complete: forgotten={forgotten}, " f"forget_failed={forget_failed}",
131 )
132
133 await _mark_job_success(session)
134
135
136 async def _mark_job_success(session) -> None:
137 """Update JobMetadata.last_success for the sweeper job."""
138 stmt = select(JobMetadata).where(JobMetadata.job_id == JOB_ID)
139 result = await session.execute(stmt)
140 job_metadata = result.scalars().first()
141 if job_metadata:
142 job_metadata.last_success = datetime.utcnow()
143 session.add(job_metadata)
144 await session.commit()
145 else:
146 logger.warning(f"JobMetadata for {JOB_ID!r} not found")