1use std::path::Path;
22
23use kimetsu_core::KimetsuResult;
24use kimetsu_core::ids::RunId;
25use rusqlite::{Connection, OptionalExtension, params};
26use time::format_description::well_known::Rfc3339;
27
28use crate::project::{load_project, load_project_readonly};
29use crate::projector;
30
31#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
38pub struct EpisodePayload {
39 pub task: String,
41 pub summary: String,
43 pub open_threads: Vec<String>,
45 pub dead_ends: Vec<String>,
47 pub hypothesis: String,
49 #[serde(default, skip_serializing_if = "str::is_empty")]
51 pub note: String,
52 pub repo_root: String,
54 #[serde(default, skip_serializing_if = "Vec::is_empty")]
56 pub memory_ids: Vec<String>,
57}
58
59#[derive(Debug, Clone)]
61pub struct EpisodeRow {
62 pub episode_id: String,
63 pub repo_root: String,
64 pub task: String,
65 pub summary: String,
66 pub open_threads: Vec<String>,
67 pub dead_ends: Vec<String>,
68 pub hypothesis: String,
69 pub note: String,
70 pub created_at: String,
71 pub superseded_by: Option<String>,
74}
75
76pub(crate) fn create_work_episodes_table(conn: &Connection) -> KimetsuResult<()> {
87 conn.execute_batch(
88 "
89 CREATE TABLE IF NOT EXISTS work_episodes (
90 episode_id TEXT PRIMARY KEY,
91 repo_root TEXT NOT NULL,
92 task TEXT NOT NULL DEFAULT '',
93 summary TEXT NOT NULL DEFAULT '',
94 open_threads TEXT NOT NULL DEFAULT '[]',
95 dead_ends TEXT NOT NULL DEFAULT '[]',
96 hypothesis TEXT NOT NULL DEFAULT '',
97 note TEXT NOT NULL DEFAULT '',
98 created_at TEXT NOT NULL,
99 superseded_by TEXT
100 );
101
102 CREATE INDEX IF NOT EXISTS idx_episodes_repo_live
103 ON work_episodes (repo_root, superseded_by);
104
105 CREATE INDEX IF NOT EXISTS idx_episodes_repo_created
106 ON work_episodes (repo_root, created_at DESC);
107 ",
108 )?;
109 Ok(())
110}
111
112pub(crate) fn project_work_episode(
124 conn: &Connection,
125 event: &kimetsu_core::event::Event,
126) -> KimetsuResult<()> {
127 let payload: EpisodePayload = serde_json::from_value(event.payload.clone()).unwrap_or_default();
128 let episode_id = event.event_id.to_string();
129 let ts = event
130 .ts
131 .format(&Rfc3339)
132 .map_err(|e| format!("format ts: {e}"))?;
133
134 let open_threads_json = serde_json::to_string(&payload.open_threads)?;
135 let dead_ends_json = serde_json::to_string(&payload.dead_ends)?;
136
137 conn.execute(
139 "
140 INSERT OR IGNORE INTO work_episodes (
141 episode_id, repo_root, task, summary, open_threads, dead_ends,
142 hypothesis, note, created_at, superseded_by
143 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, NULL)
144 ",
145 params![
146 episode_id,
147 payload.repo_root,
148 payload.task,
149 payload.summary,
150 open_threads_json,
151 dead_ends_json,
152 payload.hypothesis,
153 payload.note,
154 ts,
155 ],
156 )?;
157
158 let prior_id: Option<String> = conn
162 .query_row(
163 "
164 SELECT episode_id FROM work_episodes
165 WHERE repo_root = ?1
166 AND superseded_by IS NULL
167 AND episode_id != ?2
168 ORDER BY created_at DESC
169 LIMIT 1
170 ",
171 params![payload.repo_root, episode_id],
172 |r| r.get(0),
173 )
174 .optional()?;
175
176 if let Some(prior) = &prior_id {
177 conn.execute(
178 "UPDATE work_episodes SET superseded_by = ?2 WHERE episode_id = ?1",
179 params![prior, episode_id],
180 )?;
181 }
182
183 for memory_id in &payload.memory_ids {
188 if memory_id.is_empty() {
189 continue;
190 }
191 projector::insert_memory_edge(conn, &episode_id, memory_id, "lesson_from", &ts)?;
194 }
195
196 Ok(())
197}
198
199type EpisodeDbRow = (
205 String,
206 String,
207 String,
208 String,
209 String,
210 String,
211 String,
212 String,
213 String,
214 Option<String>,
215);
216
217pub fn load_live_episode(conn: &Connection, repo_root: &str) -> KimetsuResult<Option<EpisodeRow>> {
220 let row: Option<EpisodeDbRow> = conn
221 .query_row(
222 "
223 SELECT episode_id, repo_root, task, summary, open_threads, dead_ends,
224 hypothesis, note, created_at, superseded_by
225 FROM work_episodes
226 WHERE repo_root = ?1
227 AND superseded_by IS NULL
228 ORDER BY created_at DESC
229 LIMIT 1
230 ",
231 params![repo_root],
232 |r| {
233 Ok((
234 r.get(0)?,
235 r.get(1)?,
236 r.get(2)?,
237 r.get(3)?,
238 r.get(4)?,
239 r.get(5)?,
240 r.get(6)?,
241 r.get(7)?,
242 r.get(8)?,
243 r.get(9)?,
244 ))
245 },
246 )
247 .optional()?;
248
249 let Some((
250 episode_id,
251 repo_root_val,
252 task,
253 summary,
254 open_threads_json,
255 dead_ends_json,
256 hypothesis,
257 note,
258 created_at,
259 superseded_by,
260 )) = row
261 else {
262 return Ok(None);
263 };
264
265 let open_threads: Vec<String> = serde_json::from_str(&open_threads_json).unwrap_or_default();
266 let dead_ends: Vec<String> = serde_json::from_str(&dead_ends_json).unwrap_or_default();
267
268 Ok(Some(EpisodeRow {
269 episode_id,
270 repo_root: repo_root_val,
271 task,
272 summary,
273 open_threads,
274 dead_ends,
275 hypothesis,
276 note,
277 created_at,
278 superseded_by,
279 }))
280}
281
282pub fn render_resume_context(workspace: &Path) -> Option<String> {
304 let (paths, _config, conn) = load_project_readonly(workspace).ok()?;
305 let repo_root = paths.repo_root.to_string_lossy().to_string();
306 let episode = load_live_episode(&conn, &repo_root).ok().flatten()?;
307 Some(format_episode_for_context(&episode))
308}
309
310fn format_episode_for_context(ep: &EpisodeRow) -> String {
312 let mut parts = Vec::new();
313
314 let task = ep.task.trim();
315 if !task.is_empty() {
316 parts.push(format!(
317 "Last session in this repo: you were working on {task}."
318 ));
319 } else {
320 parts.push("Last session in this repo: previous work captured below.".to_string());
321 }
322
323 let summary = ep.summary.trim();
324 if !summary.is_empty() {
325 parts.push(format!("Done: {summary}."));
326 }
327
328 if !ep.open_threads.is_empty() {
329 let threads = ep
330 .open_threads
331 .iter()
332 .filter(|s| !s.trim().is_empty())
333 .cloned()
334 .collect::<Vec<_>>();
335 if !threads.is_empty() {
336 parts.push(format!("Open: {}.", threads.join("; ")));
337 }
338 }
339
340 if !ep.dead_ends.is_empty() {
341 let de = ep
342 .dead_ends
343 .iter()
344 .filter(|s| !s.trim().is_empty())
345 .cloned()
346 .collect::<Vec<_>>();
347 if !de.is_empty() {
348 parts.push(format!("Avoid: {}.", de.join("; ")));
349 }
350 }
351
352 let hypothesis = ep.hypothesis.trim();
353 if !hypothesis.is_empty() {
354 parts.push(format!("Hypothesis: {hypothesis}."));
355 }
356
357 if !ep.note.trim().is_empty() {
358 parts.push(format!("Note: {}.", ep.note.trim()));
359 }
360
361 parts.join("\n")
362}
363
364pub fn capture_episode(workspace: &Path, payload: EpisodePayload) -> KimetsuResult<String> {
375 let (paths, _config, conn) = load_project(workspace)?;
376 let _lock = crate::lock::ProjectLock::acquire(&paths, "episode capture", None)?;
377
378 let mut payload = payload;
382 payload.repo_root = paths.repo_root.to_string_lossy().to_string();
383
384 let run_id = RunId::new();
385 let event =
386 kimetsu_core::event::Event::new(run_id, "work.episode", serde_json::to_value(&payload)?);
387 let event_id = event.event_id.to_string();
388
389 projector::insert_event(&conn, &event)?;
391 project_work_episode(&conn, &event)?;
392
393 Ok(event_id)
394}
395
396pub fn rule_based_episode(transcript_view: &str, repo_root: &str, note: &str) -> EpisodePayload {
410 let lines: Vec<&str> = transcript_view.lines().collect();
411
412 let task = lines
414 .iter()
415 .find(|l| l.starts_with("user:"))
416 .map(|l| l.trim_start_matches("user:").trim().to_string())
417 .unwrap_or_default();
418
419 let summary = lines
421 .iter()
422 .rev()
423 .find(|l| l.starts_with("assistant:"))
424 .map(|l| l.trim_start_matches("assistant:").trim().to_string())
425 .unwrap_or_default();
426
427 let open_thread_signals = ["todo", "still need", "next step", "follow-up", "followup"];
429 let open_threads: Vec<String> = lines
430 .iter()
431 .filter(|l| {
432 let low = l.to_lowercase();
433 open_thread_signals.iter().any(|sig| low.contains(sig))
434 })
435 .take(3)
436 .map(|l| l.trim().to_string())
437 .collect();
438
439 let dead_end_signals = [
441 "failed",
442 "doesn't work",
443 "does not work",
444 "error:",
445 "gave up",
446 "won't work",
447 ];
448 let dead_ends: Vec<String> = lines
449 .iter()
450 .filter(|l| {
451 let low = l.to_lowercase();
452 dead_end_signals.iter().any(|sig| low.contains(sig))
453 })
454 .take(3)
455 .map(|l| l.trim().to_string())
456 .collect();
457
458 EpisodePayload {
459 task: task.chars().take(200).collect(),
460 summary: summary.chars().take(300).collect(),
461 open_threads,
462 dead_ends,
463 hypothesis: String::new(),
464 note: note.to_string(),
465 repo_root: repo_root.to_string(),
466 memory_ids: Vec::new(),
467 }
468}
469
470pub fn load_live_episode_for_workspace(workspace: &Path) -> KimetsuResult<Option<EpisodeRow>> {
477 let (paths, _config, conn) = load_project_readonly(workspace)?;
478 let repo_root = paths.repo_root.to_string_lossy().to_string();
479 load_live_episode(&conn, &repo_root)
480}
481
482#[cfg(test)]
487mod tests {
488 use kimetsu_core::paths::git_init_boundary;
489
490 use super::*;
491 use crate::{project, schema, user_brain};
492
493 fn tmp_workspace(name: &str) -> std::path::PathBuf {
494 let ts = std::time::SystemTime::now()
495 .duration_since(std::time::UNIX_EPOCH)
496 .map(|d| d.as_nanos())
497 .unwrap_or(0);
498 let dir = std::env::temp_dir().join(format!("kimetsu-ep-{name}-{ts}"));
499 std::fs::create_dir_all(&dir).expect("create tmp");
500 dir
501 }
502
503 fn make_in_memory_conn() -> Connection {
504 let conn = Connection::open_in_memory().expect("in-memory");
505 schema::initialize(&conn).expect("schema init");
506 conn
507 }
508
509 #[test]
513 fn project_episode_round_trip() {
514 let conn = make_in_memory_conn();
515 let run_id = kimetsu_core::ids::RunId::new();
516 let payload = EpisodePayload {
517 task: "fix the build".to_string(),
518 summary: "added missing feature flag".to_string(),
519 open_threads: vec!["still need tests".to_string()],
520 dead_ends: vec!["tried cmake, failed".to_string()],
521 hypothesis: "the linker needs explicit paths".to_string(),
522 note: "urgent".to_string(),
523 repo_root: "/repo/foo".to_string(),
524 memory_ids: vec![],
525 };
526 let event = kimetsu_core::event::Event::new(
527 run_id,
528 "work.episode",
529 serde_json::to_value(&payload).unwrap(),
530 );
531 project_work_episode(&conn, &event).expect("project_work_episode");
532
533 let ep = load_live_episode(&conn, "/repo/foo")
534 .expect("load_live_episode")
535 .expect("episode must exist");
536
537 assert_eq!(ep.task, "fix the build");
538 assert_eq!(ep.open_threads, vec!["still need tests".to_string()]);
539 assert_eq!(ep.dead_ends, vec!["tried cmake, failed".to_string()]);
540 assert_eq!(ep.hypothesis, "the linker needs explicit paths");
541 assert_eq!(ep.note, "urgent");
542 assert!(
543 ep.superseded_by.is_none(),
544 "new episode must not be superseded"
545 );
546 }
547
548 #[test]
552 fn second_episode_supersedes_first() {
553 let conn = make_in_memory_conn();
554 let run_id = kimetsu_core::ids::RunId::new();
555
556 let payload1 = EpisodePayload {
557 task: "task 1".to_string(),
558 repo_root: "/repo/bar".to_string(),
559 ..Default::default()
560 };
561 let ev1 = kimetsu_core::event::Event::new(
562 run_id,
563 "work.episode",
564 serde_json::to_value(&payload1).unwrap(),
565 );
566 project_work_episode(&conn, &ev1).expect("project ev1");
567
568 let ep1_id = ev1.event_id.to_string();
569
570 let mut ev2 = kimetsu_core::event::Event::new(
574 run_id,
575 "work.episode",
576 serde_json::to_value(EpisodePayload {
577 task: "task 2".to_string(),
578 repo_root: "/repo/bar".to_string(),
579 ..Default::default()
580 })
581 .unwrap(),
582 );
583 ev2.ts = ev1.ts + time::Duration::seconds(1);
585
586 project_work_episode(&conn, &ev2).expect("project ev2");
587
588 let ep1: Option<String> = conn
590 .query_row(
591 "SELECT superseded_by FROM work_episodes WHERE episode_id = ?1",
592 [ep1_id],
593 |r| r.get(0),
594 )
595 .expect("query ep1");
596 assert!(
597 ep1.is_some(),
598 "first episode must be superseded after second"
599 );
600
601 let live = load_live_episode(&conn, "/repo/bar")
603 .expect("load")
604 .expect("live episode");
605 assert_eq!(live.task, "task 2");
606 assert!(live.superseded_by.is_none());
607 }
608
609 #[test]
613 fn reset_projection_clears_episodes() {
614 let conn = make_in_memory_conn();
615 let run_id = kimetsu_core::ids::RunId::new();
616 let payload = EpisodePayload {
617 task: "some task".to_string(),
618 repo_root: "/repo/reset".to_string(),
619 ..Default::default()
620 };
621 let event = kimetsu_core::event::Event::new(
622 run_id,
623 "work.episode",
624 serde_json::to_value(&payload).unwrap(),
625 );
626 crate::projector::apply_events(&conn, &[event]).expect("apply_events");
628
629 let count_before: i64 = conn
630 .query_row("SELECT COUNT(*) FROM work_episodes", [], |r| r.get(0))
631 .unwrap();
632 assert_eq!(count_before, 1, "episode must exist before reset");
633
634 conn.execute_batch(
636 "DELETE FROM runs; DELETE FROM sources; DELETE FROM memories; \
637 DELETE FROM memory_proposals; DELETE FROM memories_fts; DELETE FROM memory_citations; \
638 DELETE FROM memory_conflicts; DELETE FROM memory_edges; DELETE FROM work_episodes;",
639 )
640 .expect("manual reset");
641
642 let count_after: i64 = conn
643 .query_row("SELECT COUNT(*) FROM work_episodes", [], |r| r.get(0))
644 .unwrap();
645 assert_eq!(count_after, 0, "work_episodes must be cleared after reset");
646 }
647
648 #[test]
652 fn rebuild_in_place_reprojects_episodes() {
653 let conn = make_in_memory_conn();
654 let run_id = kimetsu_core::ids::RunId::new();
655 let payload = EpisodePayload {
656 task: "rebuild test".to_string(),
657 repo_root: "/repo/rebuild".to_string(),
658 ..Default::default()
659 };
660 let event = kimetsu_core::event::Event::new(
661 run_id,
662 "work.episode",
663 serde_json::to_value(&payload).unwrap(),
664 );
665 crate::projector::apply_events(&conn, &[event]).expect("apply_events");
666
667 conn.execute("DELETE FROM work_episodes", []).unwrap();
669 let count_wiped: i64 = conn
670 .query_row("SELECT COUNT(*) FROM work_episodes", [], |r| r.get(0))
671 .unwrap();
672 assert_eq!(count_wiped, 0, "work_episodes wiped before rebuild");
673
674 let replayed = crate::projector::rebuild_in_place(&conn).expect("rebuild_in_place");
676 assert!(replayed >= 1, "at least 1 event replayed");
677
678 let ep = load_live_episode(&conn, "/repo/rebuild")
679 .expect("load")
680 .expect("episode restored");
681 assert_eq!(ep.task, "rebuild test");
682 }
683
684 #[test]
688 fn render_resume_context_formats_episode() {
689 let ep = EpisodeRow {
690 episode_id: "ep1".to_string(),
691 repo_root: "/r".to_string(),
692 task: "implement feature X".to_string(),
693 summary: "added the core logic".to_string(),
694 open_threads: vec!["write tests".to_string(), "update docs".to_string()],
695 dead_ends: vec!["tried approach A, OOM".to_string()],
696 hypothesis: "batching is the fix".to_string(),
697 note: String::new(),
698 created_at: "2026-01-01T00:00:00Z".to_string(),
699 superseded_by: None,
700 };
701 let text = format_episode_for_context(&ep);
702 assert!(text.contains("implement feature X"), "task in output");
703 assert!(text.contains("added the core logic"), "summary in output");
704 assert!(text.contains("write tests"), "open thread in output");
705 assert!(text.contains("tried approach A"), "dead-end in output");
706 assert!(text.contains("batching is the fix"), "hypothesis in output");
707 assert!(
709 text.len() < 800,
710 "render output must be concise (< 800 chars), got {}",
711 text.len()
712 );
713 }
714
715 #[test]
719 fn rule_based_episode_parses_transcript() {
720 let view = "user: implement the auth module\nassistant: I need to still need integration tests\n\
721 assistant: cargo build failed — error: linker not found\nassistant: implemented the core auth flow.";
722 let ep = rule_based_episode(view, "/r", "my note");
723 assert_eq!(ep.task, "implement the auth module");
724 assert!(!ep.summary.is_empty(), "summary extracted");
725 assert_eq!(ep.note, "my note");
726 assert_eq!(ep.repo_root, "/r");
727 }
728
729 #[test]
733 fn capture_episode_end_to_end() {
734 let dir = tmp_workspace("capture");
735 git_init_boundary(&dir);
736 user_brain::with_user_brain_disabled(|| {
737 project::init_project(&dir, true).expect("init");
738 let payload = EpisodePayload {
739 task: "e2e task".to_string(),
740 summary: "e2e summary".to_string(),
741 repo_root: dir.to_string_lossy().to_string(),
742 ..Default::default()
743 };
744 let id = capture_episode(&dir, payload).expect("capture_episode");
745 assert!(!id.is_empty(), "episode_id returned");
746
747 let ep = load_live_episode_for_workspace(&dir)
748 .expect("load")
749 .expect("live episode");
750 assert_eq!(ep.task, "e2e task");
751 });
752 std::fs::remove_dir_all(dir).ok();
753 }
754
755 #[test]
759 fn episode_inserts_lesson_from_edges() {
760 let conn = make_in_memory_conn();
761 let run_id = kimetsu_core::ids::RunId::new();
762 let payload = EpisodePayload {
763 task: "edge test".to_string(),
764 repo_root: "/repo/edges".to_string(),
765 memory_ids: vec!["mem-abc".to_string(), "mem-xyz".to_string()],
766 ..Default::default()
767 };
768 let event = kimetsu_core::event::Event::new(
769 run_id,
770 "work.episode",
771 serde_json::to_value(&payload).unwrap(),
772 );
773 let event_id_str = event.event_id.to_string();
774 crate::projector::apply_events(&conn, std::slice::from_ref(&event)).expect("apply_events");
775
776 let edge_count: i64 = conn
777 .query_row(
778 "SELECT COUNT(*) FROM memory_edges WHERE src_id = ?1 AND edge_type = 'lesson_from'",
779 [event_id_str],
780 |r| r.get(0),
781 )
782 .expect("query edges");
783 assert_eq!(edge_count, 2, "two lesson_from edges inserted");
784 }
785}