1use std::collections::HashSet;
2
3use anyhow::{bail, Context, Result};
4use rusqlite::{named_params, params, Connection, OptionalExtension, TransactionBehavior};
5use serde::Serialize;
6
7use crate::db::{self, CaptureEventInput, ExtractionTaskKind};
8
9use super::bridge_state::AUTO_ACTIONABLE_PREDICATE;
10
11const AUTO_MIGRATION_RETRY_BASE_SECS: i64 = 5;
12const AUTO_MIGRATION_RETRY_MAX_SECS: i64 = 900;
13const AUTO_MIGRATION_RETRY_MAX_SHIFT: i64 = 8;
14
15const MANUAL_ELIGIBLE_PREDICATE: &str = "
16 (status = 'pending'
17 OR (status = 'processing'
18 AND (lease_expires_epoch IS NULL OR lease_expires_epoch < :now)))";
19
20const LEGACY_PENDING_SNAPSHOT_COLUMNS: &str = "
21 id, host, session_id, project, tool_name, tool_input, tool_response, cwd,
22 created_at_epoch, updated_at_epoch, status, attempt_count, next_retry_epoch,
23 last_error, lease_owner, lease_expires_epoch, failure_class, failed_at_epoch,
24 archived_at_epoch";
25
26#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
27pub struct LegacyPendingMigration {
28 pub pending_id: i64,
29 pub event_id: String,
30 pub captured_event_id: i64,
31 pub extraction_task_id: i64,
32 pub host: String,
33 pub project: String,
34 pub session_id: String,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub(super) struct LegacyPendingRow {
39 pub(super) id: i64,
40 pub(super) host: String,
41 pub(super) session_id: String,
42 pub(super) project: String,
43 pub(super) tool_name: String,
44 pub(super) tool_input: Option<String>,
45 pub(super) tool_response: Option<String>,
46 pub(super) cwd: Option<String>,
47 pub(super) created_at_epoch: i64,
48}
49
50pub(super) struct PreparedLegacyReplay {
51 snapshot: LegacyPendingRow,
52 content: String,
53 git_branch: Option<String>,
54}
55
56#[derive(PartialEq, Eq)]
57struct LegacyPendingSnapshot {
58 legacy: LegacyPendingRow,
59 updated_at_epoch: i64,
60 status: String,
61 attempt_count: i64,
62 next_retry_epoch: Option<i64>,
63 last_error: Option<String>,
64 lease_owner: Option<String>,
65 lease_expires_epoch: Option<i64>,
66 failure_class: Option<String>,
67 failed_at_epoch: Option<i64>,
68 archived_at_epoch: Option<i64>,
69}
70
71struct PreparedManualLegacyReplay {
72 source: LegacyPendingSnapshot,
73 replay: PreparedLegacyReplay,
74 host: String,
75}
76
77impl PreparedLegacyReplay {
78 pub(super) fn matches(&self, row: &LegacyPendingRow) -> bool {
79 &self.snapshot == row
80 }
81}
82
83pub fn count_legacy_migration_candidates(
84 conn: &Connection,
85 project: Option<&str>,
86 limit: i64,
87) -> Result<usize> {
88 let limit = limit.max(1);
89 let now = chrono::Utc::now().timestamp();
90 let count: i64 = if let Some(project) = project {
91 conn.query_row(
92 "SELECT COUNT(*) FROM (
93 SELECT id FROM pending_observations
94 WHERE project = ?1
95 AND (status = 'pending'
96 OR (status = 'processing'
97 AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?3)))
98 ORDER BY created_at_epoch ASC, id ASC
99 LIMIT ?2
100 )",
101 params![project, limit, now],
102 |row| row.get(0),
103 )?
104 } else {
105 conn.query_row(
106 "SELECT COUNT(*) FROM (
107 SELECT id FROM pending_observations
108 WHERE status = 'pending'
109 OR (status = 'processing'
110 AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?2))
111 ORDER BY created_at_epoch ASC, id ASC
112 LIMIT ?1
113 )",
114 params![limit, now],
115 |row| row.get(0),
116 )?
117 };
118 Ok(count.max(0) as usize)
119}
120
121pub fn count_recoverable_archived_legacy_pending(conn: &Connection) -> Result<usize> {
122 Ok(super::query::query_archived_transient_legacy_pending(conn)?.due)
123}
124
125pub fn count_admin_required_archived_legacy_pending(conn: &Connection) -> Result<usize> {
126 let count: i64 = conn.query_row(
127 "SELECT COUNT(*)
128 FROM pending_observations
129 WHERE status = 'failed'
130 AND archived_at_epoch IS NOT NULL
131 AND NOT (
132 host IN (?1, ?2)
133 AND COALESCE(failure_class, 'transient') = 'transient'
134 )",
135 params![
136 crate::runtime_config::CLAUDE_HOST,
137 crate::runtime_config::CODEX_HOST
138 ],
139 |row| row.get(0),
140 )?;
141 Ok(count.max(0) as usize)
142}
143
144pub fn migrate_legacy_pending(
145 conn: &mut Connection,
146 project: Option<&str>,
147 fallback_host: Option<&str>,
148 limit: i64,
149) -> Result<Vec<LegacyPendingMigration>> {
150 let mut detector = db::detect_git_branch;
151 migrate_legacy_pending_with_detector(conn, project, fallback_host, limit, &mut detector)
152}
153
154fn migrate_legacy_pending_with_detector(
155 conn: &mut Connection,
156 project: Option<&str>,
157 fallback_host: Option<&str>,
158 limit: i64,
159 detector: &mut dyn FnMut(&str) -> Option<String>,
160) -> Result<Vec<LegacyPendingMigration>> {
161 let fallback_host = fallback_host.map(normalize_capture_host).transpose()?;
162 let rows = select_legacy_pending_rows(conn, project, limit)?;
163 let prepared = rows
164 .into_iter()
165 .map(|source| {
166 let host = capture_host_for_row(&source.legacy.host, fallback_host)?.to_string();
167 let replay = prepare_legacy_replay_with_detector(&source.legacy, detector);
168 Ok(PreparedManualLegacyReplay {
169 source,
170 replay,
171 host,
172 })
173 })
174 .collect::<Result<Vec<_>>>()?;
175 if prepared.is_empty() {
176 return Ok(Vec::new());
177 }
178
179 let tx = conn
180 .transaction_with_behavior(TransactionBehavior::Immediate)
181 .context("begin manual legacy pending migration transaction")?;
182 let mut migrated = Vec::new();
183
184 for prepared_row in prepared {
185 let pending_id = prepared_row.source.legacy.id;
186 let eligibility_now = chrono::Utc::now().timestamp();
187 let row = load_legacy_pending_row(&tx, pending_id, eligibility_now)?.ok_or_else(|| {
188 anyhow::anyhow!(
189 "legacy pending row {pending_id} changed or became ineligible while preparing migration; batch replay was rolled back"
190 )
191 })?;
192 if prepared_row.source != row || !prepared_row.replay.matches(&row.legacy) {
193 bail!(
194 "legacy pending row {pending_id} changed while preparing migration; batch replay was rolled back"
195 );
196 }
197 let migration =
198 replay_prepared_legacy_row_into_capture(&tx, &prepared_row.replay, &prepared_row.host)
199 .with_context(|| format!("replay legacy pending row {pending_id}"))?;
200 let completed_at = chrono::Utc::now().timestamp();
201 let changed = tx.execute(
202 MARK_MIGRATED_PENDING_SQL,
203 params![pending_id, completed_at, eligibility_now],
204 )?;
205 if changed != 1 {
206 bail!(
207 "legacy pending row {pending_id} changed while migrating; batch replay was rolled back"
208 );
209 }
210 migrated.push(migration);
211 }
212
213 tx.commit()
214 .context("commit manual legacy pending migration")?;
215 Ok(migrated)
216}
217
218const MARK_MIGRATED_PENDING_SQL: &str = "UPDATE pending_observations
219 SET status = 'migrated',
220 lease_owner = NULL,
221 lease_expires_epoch = NULL,
222 next_retry_epoch = NULL,
223 last_error = NULL,
224 updated_at_epoch = ?2
225 WHERE id = ?1
226 AND (status = 'pending'
227 OR (status = 'processing'
228 AND (lease_expires_epoch IS NULL OR lease_expires_epoch < ?3)))";
229
230pub(super) fn prepare_legacy_replay_with_detector(
231 row: &LegacyPendingRow,
232 detector: &mut dyn FnMut(&str) -> Option<String>,
233) -> PreparedLegacyReplay {
234 let git_branch = row.cwd.as_deref().and_then(detector);
235 let content = legacy_capture_content(row, git_branch.as_deref());
236 PreparedLegacyReplay {
237 snapshot: row.clone(),
238 content,
239 git_branch,
240 }
241}
242
243pub(super) fn replay_prepared_legacy_row_into_capture(
244 conn: &Connection,
245 prepared: &PreparedLegacyReplay,
246 host: &str,
247) -> Result<LegacyPendingMigration> {
248 let row = &prepared.snapshot;
249 let event_id = legacy_event_id(row.id);
250 let outcome =
251 db::capture::record_captured_event_with_id_and_created_at_and_precomputed_git_branch(
252 conn,
253 &CaptureEventInput {
254 host,
255 session_id: &row.session_id,
256 project: &row.project,
257 cwd: row.cwd.as_deref(),
258 event_type: "tool_result",
259 role: None,
260 tool_name: Some(&row.tool_name),
261 content: &prepared.content,
262 task_kind: Some(ExtractionTaskKind::ObservationExtract),
263 },
264 Some(&event_id),
265 row.created_at_epoch,
266 prepared.git_branch.as_deref(),
267 )?;
268 let extraction_task_id = outcome
269 .extraction_task_id
270 .ok_or_else(|| anyhow::anyhow!("legacy pending migration did not enqueue extraction"))?;
271 Ok(LegacyPendingMigration {
272 pending_id: row.id,
273 event_id,
274 captured_event_id: outcome.event_row_id,
275 extraction_task_id,
276 host: host.to_string(),
277 project: row.project.clone(),
278 session_id: row.session_id.clone(),
279 })
280}
281
282#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
283pub struct AutoLegacyMigrationOutcome {
284 pub migrated: usize,
285 pub yielded_to_current_work: bool,
286}
287
288#[derive(Clone, Copy, Debug, Eq, PartialEq)]
289enum AutoCandidateOutcome {
290 Migrated {
291 extraction_task_id: i64,
292 captured_event_id: i64,
293 },
294 Skipped,
295 YieldedToCurrentWork,
296}
297
298#[derive(Clone, Copy, Debug, Eq, PartialEq)]
299struct AutoMigrationRetry {
300 attempt_count: i64,
301 next_retry_epoch: i64,
302 backoff_secs: i64,
303}
304
305pub fn auto_migrate_actionable_legacy_pending(
315 conn: &mut Connection,
316 limit: i64,
317) -> Result<AutoLegacyMigrationOutcome> {
318 let mut detector = db::detect_git_branch;
319 auto_migrate_actionable_legacy_pending_with_detector(conn, limit, &mut detector)
320}
321
322pub(super) fn auto_migrate_actionable_legacy_pending_with_detector(
323 conn: &mut Connection,
324 limit: i64,
325 detector: &mut dyn FnMut(&str) -> Option<String>,
326) -> Result<AutoLegacyMigrationOutcome> {
327 let candidate_ids = select_auto_actionable_ids(conn, limit)?;
328 let mut outcome = AutoLegacyMigrationOutcome::default();
329 let mut migrated_tasks = HashSet::new();
330 for row_id in candidate_ids {
331 if current_extraction_work_is_ready(conn, &migrated_tasks)? {
332 outcome.yielded_to_current_work = true;
333 break;
334 }
335 match auto_migrate_candidate_with_detector(
336 conn,
337 row_id,
338 detector,
339 &migrated_tasks,
340 ) {
341 Ok(AutoCandidateOutcome::Migrated {
342 extraction_task_id,
343 captured_event_id,
344 }) => {
345 outcome.migrated += 1;
346 migrated_tasks.insert((extraction_task_id, captured_event_id));
347 }
348 Ok(AutoCandidateOutcome::Skipped) => {}
349 Ok(AutoCandidateOutcome::YieldedToCurrentWork) => {
350 outcome.yielded_to_current_work = true;
351 break;
352 }
353 Err(error) => {
354 return Err(error).with_context(|| {
355 format!(
356 "legacy pending auto-migration aborted batch id={row_id} migrated_before_error={}",
357 outcome.migrated
358 )
359 })
360 }
361 }
362 }
363 Ok(outcome)
364}
365
366fn current_extraction_work_is_ready(
367 conn: &Connection,
368 migrated_tasks: &HashSet<(i64, i64)>,
369) -> Result<bool> {
370 let now = chrono::Utc::now().timestamp();
371 let mut stmt = conn.prepare(
372 "SELECT id, status, high_watermark_event_id, next_retry_epoch
373 FROM extraction_tasks
374 WHERE (status = 'pending'
375 AND (next_retry_epoch IS NULL OR next_retry_epoch <= ?1))
376 OR status = 'processing'",
377 )?;
378 let rows = stmt.query_map([now], |row| {
379 Ok((
380 row.get::<_, i64>(0)?,
381 row.get::<_, String>(1)?,
382 row.get::<_, Option<i64>>(2)?,
383 row.get::<_, Option<i64>>(3)?,
384 ))
385 })?;
386 for task in rows {
387 match task? {
388 (task_id, status, Some(event_id), None)
389 if status == "pending" && migrated_tasks.contains(&(task_id, event_id)) => {}
390 _ => return Ok(true),
391 }
392 }
393 Ok(false)
394}
395
396fn select_auto_actionable_ids(conn: &Connection, limit: i64) -> Result<Vec<i64>> {
397 let limit = limit.max(1);
398 let now = chrono::Utc::now().timestamp();
399 let sql = format!(
400 "SELECT id
401 FROM pending_observations
402 WHERE {AUTO_ACTIONABLE_PREDICATE}
403 ORDER BY created_at_epoch ASC, id ASC
404 LIMIT :limit"
405 );
406 let mut stmt = conn.prepare(&sql)?;
407 let rows = stmt.query_map(
408 named_params! {
409 ":now": now,
410 ":claude_host": crate::runtime_config::CLAUDE_HOST,
411 ":codex_host": crate::runtime_config::CODEX_HOST,
412 ":limit": limit,
413 },
414 |row| row.get(0),
415 )?;
416 rows.collect::<std::result::Result<Vec<_>, _>>()
417 .map_err(Into::into)
418}
419
420fn auto_migrate_candidate_with_detector(
421 conn: &mut Connection,
422 row_id: i64,
423 detector: &mut dyn FnMut(&str) -> Option<String>,
424 migrated_tasks: &HashSet<(i64, i64)>,
425) -> Result<AutoCandidateOutcome> {
426 let preflight_now = chrono::Utc::now().timestamp();
427 let Some(preflight_row) = load_auto_actionable_row(conn, row_id, preflight_now)? else {
428 return Ok(AutoCandidateOutcome::Skipped);
429 };
430 normalize_capture_host(&preflight_row.legacy.host)?;
431 let prepared = prepare_legacy_replay_with_detector(&preflight_row.legacy, detector);
432 if current_extraction_work_is_ready(conn, migrated_tasks)? {
433 return Ok(AutoCandidateOutcome::YieldedToCurrentWork);
434 }
435
436 let mut tx = conn
437 .transaction_with_behavior(TransactionBehavior::Immediate)
438 .context("begin legacy pending auto-migration transaction")?;
439 if current_extraction_work_is_ready(&tx, migrated_tasks)? {
440 tx.commit()?;
441 return Ok(AutoCandidateOutcome::YieldedToCurrentWork);
442 }
443 let eligibility_now = chrono::Utc::now().timestamp();
444 let Some(row) = load_auto_actionable_row(&tx, row_id, eligibility_now)? else {
445 tx.commit()?;
446 return Ok(AutoCandidateOutcome::Skipped);
447 };
448 if preflight_row != row || !prepared.matches(&row.legacy) {
449 tx.commit()?;
450 return Ok(AutoCandidateOutcome::Skipped);
451 }
452 let host = normalize_capture_host(&row.legacy.host)?;
453 let replay = {
454 let savepoint = tx
455 .savepoint_with_name("legacy_pending_auto_replay")
456 .context("begin legacy pending replay savepoint")?;
457 let replay = replay_prepared_legacy_row_into_capture(&savepoint, &prepared, host);
458 match replay {
459 Ok(migration) => {
460 savepoint
461 .commit()
462 .context("commit legacy pending replay savepoint")?;
463 Ok(migration)
464 }
465 Err(error) => {
466 savepoint
467 .finish()
468 .context("roll back legacy pending replay savepoint")?;
469 Err(error)
470 }
471 }
472 };
473 let migration = match replay {
474 Ok(migration) => migration,
475 Err(error) => {
476 let retry = mark_legacy_row_for_transient_retry(
477 &tx,
478 row_id,
479 eligibility_now,
480 &format!("{error:#}"),
481 )
482 .with_context(|| format!("record legacy pending retry state id={row_id}"))?
483 .ok_or_else(|| {
484 anyhow::anyhow!(
485 "legacy pending row changed inside immediate transaction id={row_id}"
486 )
487 })?;
488 tx.commit()
489 .context("commit legacy pending retry transition")?;
490 crate::log::error(
491 "failure_lifecycle",
492 &format!(
493 "surface=pending_observation class=transient outcome=deferred id={row_id} attempt={} backoff_secs={} next_retry_epoch={}",
494 retry.attempt_count, retry.backoff_secs, retry.next_retry_epoch
495 ),
496 );
497 return Err(error).with_context(|| {
498 format!(
499 "legacy pending auto-migration scheduled retry id={row_id} attempt={} backoff_secs={}",
500 retry.attempt_count, retry.backoff_secs
501 )
502 });
503 }
504 };
505
506 let completed_at = chrono::Utc::now().timestamp();
507 let changed = mark_auto_migrated(&tx, row_id, eligibility_now, completed_at)?;
508 match changed {
509 0 => {
510 tx.rollback()?;
511 Ok(AutoCandidateOutcome::Skipped)
512 }
513 1 => {
514 tx.commit()?;
515 Ok(AutoCandidateOutcome::Migrated {
516 extraction_task_id: migration.extraction_task_id,
517 captured_event_id: migration.captured_event_id,
518 })
519 }
520 changed => bail!(
521 "legacy pending auto-migration invariant violated: id={row_id} affected_rows={changed}"
522 ),
523 }
524}
525
526fn load_auto_actionable_row(
527 conn: &Connection,
528 row_id: i64,
529 now: i64,
530) -> Result<Option<LegacyPendingSnapshot>> {
531 let sql = format!(
532 "SELECT {LEGACY_PENDING_SNAPSHOT_COLUMNS}
533 FROM pending_observations
534 WHERE id = :id
535 AND {AUTO_ACTIONABLE_PREDICATE}"
536 );
537 conn.query_row(
538 &sql,
539 named_params! {
540 ":id": row_id,
541 ":now": now,
542 ":claude_host": crate::runtime_config::CLAUDE_HOST,
543 ":codex_host": crate::runtime_config::CODEX_HOST,
544 },
545 snapshot_from_db,
546 )
547 .optional()
548 .map_err(Into::into)
549}
550
551fn mark_auto_migrated(
552 conn: &Connection,
553 row_id: i64,
554 eligibility_now: i64,
555 completed_at: i64,
556) -> Result<usize> {
557 let sql = format!(
558 "UPDATE pending_observations
559 SET status = 'migrated',
560 attempt_count = 0,
561 lease_owner = NULL,
562 lease_expires_epoch = NULL,
563 next_retry_epoch = NULL,
564 last_error = NULL,
565 failure_class = NULL,
566 failed_at_epoch = NULL,
567 archived_at_epoch = NULL,
568 updated_at_epoch = :completed_at
569 WHERE id = :id
570 AND {AUTO_ACTIONABLE_PREDICATE}"
571 );
572 Ok(conn.execute(
573 &sql,
574 named_params! {
575 ":id": row_id,
576 ":now": eligibility_now,
577 ":completed_at": completed_at,
578 ":claude_host": crate::runtime_config::CLAUDE_HOST,
579 ":codex_host": crate::runtime_config::CODEX_HOST,
580 },
581 )?)
582}
583
584fn mark_legacy_row_for_transient_retry(
585 conn: &Connection,
586 row_id: i64,
587 now: i64,
588 error: &str,
589) -> Result<Option<AutoMigrationRetry>> {
590 let marker = format!(
591 "[auto_migration_retry] {}",
592 crate::db::truncate_str(error, 1000)
593 );
594 let sql = format!(
595 "UPDATE pending_observations
596 SET status = 'failed',
597 attempt_count = attempt_count + 1,
598 next_retry_epoch = :now + MIN(
599 :max_retry_secs,
600 :base_retry_secs * (1 << MIN(MAX(COALESCE(attempt_count, 0), 0), :max_shift))
601 ),
602 lease_owner = NULL,
603 lease_expires_epoch = NULL,
604 failure_class = 'transient',
605 failed_at_epoch = COALESCE(failed_at_epoch, :now),
606 last_error = :error,
607 updated_at_epoch = :now
608 WHERE id = :id
609 AND {AUTO_ACTIONABLE_PREDICATE}"
610 );
611 let changed = conn.execute(
612 &sql,
613 named_params! {
614 ":id": row_id,
615 ":now": now,
616 ":claude_host": crate::runtime_config::CLAUDE_HOST,
617 ":codex_host": crate::runtime_config::CODEX_HOST,
618 ":base_retry_secs": AUTO_MIGRATION_RETRY_BASE_SECS,
619 ":max_retry_secs": AUTO_MIGRATION_RETRY_MAX_SECS,
620 ":max_shift": AUTO_MIGRATION_RETRY_MAX_SHIFT,
621 ":error": marker,
622 },
623 )?;
624 if changed == 0 {
625 return Ok(None);
626 }
627 let (attempt_count, next_retry_epoch): (i64, i64) = conn.query_row(
628 "SELECT attempt_count, next_retry_epoch
629 FROM pending_observations
630 WHERE id = ?1",
631 params![row_id],
632 |row| Ok((row.get(0)?, row.get(1)?)),
633 )?;
634 Ok(Some(AutoMigrationRetry {
635 attempt_count,
636 next_retry_epoch,
637 backoff_secs: next_retry_epoch.saturating_sub(now),
638 }))
639}
640
641fn select_legacy_pending_rows(
642 conn: &Connection,
643 project: Option<&str>,
644 limit: i64,
645) -> Result<Vec<LegacyPendingSnapshot>> {
646 let limit = limit.max(1);
647 let now = chrono::Utc::now().timestamp();
648 let sql = format!(
649 "SELECT {LEGACY_PENDING_SNAPSHOT_COLUMNS}
650 FROM pending_observations
651 WHERE (:project IS NULL OR project = :project)
652 AND {MANUAL_ELIGIBLE_PREDICATE}
653 ORDER BY created_at_epoch ASC, id ASC
654 LIMIT :limit"
655 );
656 let mut stmt = conn.prepare(&sql)?;
657 let rows = stmt.query_map(
658 named_params! {
659 ":project": project,
660 ":now": now,
661 ":limit": limit,
662 },
663 snapshot_from_db,
664 )?;
665 rows.collect::<std::result::Result<Vec<_>, _>>()
666 .map_err(Into::into)
667}
668
669fn load_legacy_pending_row(
670 conn: &Connection,
671 pending_id: i64,
672 now: i64,
673) -> Result<Option<LegacyPendingSnapshot>> {
674 let sql = format!(
675 "SELECT {LEGACY_PENDING_SNAPSHOT_COLUMNS}
676 FROM pending_observations
677 WHERE id = :id
678 AND {MANUAL_ELIGIBLE_PREDICATE}"
679 );
680 conn.query_row(
681 &sql,
682 named_params! {
683 ":id": pending_id,
684 ":now": now,
685 },
686 snapshot_from_db,
687 )
688 .optional()
689 .map_err(Into::into)
690}
691
692fn snapshot_from_db(row: &rusqlite::Row<'_>) -> rusqlite::Result<LegacyPendingSnapshot> {
693 Ok(LegacyPendingSnapshot {
694 legacy: row_from_db(row)?,
695 updated_at_epoch: row.get(9)?,
696 status: row.get(10)?,
697 attempt_count: row.get(11)?,
698 next_retry_epoch: row.get(12)?,
699 last_error: row.get(13)?,
700 lease_owner: row.get(14)?,
701 lease_expires_epoch: row.get(15)?,
702 failure_class: row.get(16)?,
703 failed_at_epoch: row.get(17)?,
704 archived_at_epoch: row.get(18)?,
705 })
706}
707
708fn row_from_db(row: &rusqlite::Row<'_>) -> rusqlite::Result<LegacyPendingRow> {
709 Ok(LegacyPendingRow {
710 id: row.get(0)?,
711 host: row.get(1)?,
712 session_id: row.get(2)?,
713 project: row.get(3)?,
714 tool_name: row.get(4)?,
715 tool_input: row.get(5)?,
716 tool_response: row.get(6)?,
717 cwd: row.get(7)?,
718 created_at_epoch: row.get(8)?,
719 })
720}
721
722fn capture_host_for_row<'a>(row_host: &'a str, fallback_host: Option<&'a str>) -> Result<&'a str> {
723 match normalize_capture_host(row_host) {
724 Ok(host) => Ok(host),
725 Err(_) => fallback_host
726 .ok_or_else(|| anyhow::anyhow!("legacy pending row has host='{row_host}'; pass --host claude-code or --host codex-cli")),
727 }
728}
729
730fn normalize_capture_host(host: &str) -> Result<&str> {
731 match host {
732 crate::runtime_config::CLAUDE_HOST | crate::runtime_config::CODEX_HOST => Ok(host),
733 _ => bail!("invalid capture host '{host}'"),
734 }
735}
736
737fn legacy_event_id(id: i64) -> String {
738 format!("legacy-pending-{id}")
739}
740
741fn legacy_capture_content(row: &LegacyPendingRow, git_branch: Option<&str>) -> String {
742 serde_json::json!({
743 "summary": format!("Recovered legacy {} event", row.tool_name),
744 "event_type": "legacy_pending_observation",
745 "detail": format!(
746 "Recovered from pending_observations id={} created_at_epoch={}",
747 row.id, row.created_at_epoch
748 ),
749 "files": serde_json::Value::Null,
750 "exit_code": serde_json::Value::Null,
751 "tool_name": row.tool_name,
752 "tool_input": parse_jsonish(row.tool_input.as_deref()),
753 "tool_response": parse_jsonish(row.tool_response.as_deref()),
754 "git_branch": git_branch,
755 })
756 .to_string()
757}
758
759fn parse_jsonish(value: Option<&str>) -> serde_json::Value {
760 match value {
761 Some(value) => serde_json::from_str(value)
762 .unwrap_or_else(|_| serde_json::Value::String(value.to_string())),
763 None => serde_json::Value::Null,
764 }
765}
766
767#[cfg(test)]
768mod tests;