1use std::{
14 collections::{BTreeSet, HashMap, HashSet},
15 path::{Path, PathBuf},
16};
17
18use anyhow::Result;
19use chrono::Utc;
20use objects::{
21 object::{StateId, ThreadName, Tree},
22 worktree::WorktreeStatus,
23};
24use oplog::{OpLogRecorder, OpRecord, ThreadUpdateSnapshots};
25use refs::{RefExpectation, RefUpdate};
26use repo::{
27 ActorPresence, ActorPresenceStatus, ActorPresenceStore, AgentTaskRecord, AgentTaskStore,
28 AgentUsageSummary, GitOverlayBranchTip, GitRemoteTrackingStatus, Repository,
29 RepositoryOperationStatus, Thread, ThreadConfidenceSummary, ThreadFreshness,
30 ThreadImpactCategory, ThreadIntegrationPolicy, ThreadManager, ThreadMode, ThreadRuntimeOverlay,
31 ThreadState, ThreadVerificationSummary, ThreadView, describe_thread_advice,
32 refresh_thread_freshness, shell_quote,
33};
34use schemars::JsonSchema;
35use serde::Serialize;
36use sley::Repository as SleyRepository;
37
38use crate::{
39 ActionTemplate,
40 status::{
41 CoordinationStatus,
42 next_action::{
43 NextActionInput, canonical_git_repair_ref_preview_command, contextual_thread_action,
44 effective_next_action, heddle_action,
45 },
46 },
47 verify::{action_template, serialize_empty_action_as_null},
48};
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
52pub struct ThreadListOptions {
53 pub include_auto: bool,
56 pub include_abandoned: bool,
59}
60
61pub struct ThreadUpdateBefore {
68 pub state: StateId,
69 ref_absent: bool,
70 manager_snapshot: Option<Vec<u8>>,
71 manager_records: Vec<Thread>,
72}
73
74pub fn capture_thread_update_before(
76 repo: &Repository,
77 manager: &ThreadManager,
78 thread: &Thread,
79) -> Result<ThreadUpdateBefore> {
80 let thread_name = ThreadName::new(&thread.thread);
81 let (state, ref_absent) = match repo.refs().get_thread(&thread_name)? {
82 Some(state) => (state, false),
83 None => {
84 let state = if let Some(current) = thread.current_state.as_deref() {
85 repo.resolve_state(current)?
86 } else {
87 repo.resolve_state(&thread.base_state)?
88 }
89 .ok_or_else(|| anyhow::anyhow!("state not found for thread '{}'", thread.thread))?;
90 (state, true)
91 }
92 };
93 Ok(ThreadUpdateBefore {
94 state,
95 ref_absent,
96 manager_snapshot: manager.snapshot_thread_record(&thread.thread)?,
97 manager_records: manager.snapshot_records(&thread.thread)?,
98 })
99}
100
101pub fn save_thread_update(
103 repo: &Repository,
104 manager: &ThreadManager,
105 thread: &Thread,
106 before: ThreadUpdateBefore,
107 new_state: StateId,
108) -> Result<()> {
109 let new_manager_snapshot = Some(manager.encode_thread_record_snapshot(thread)?);
110 let mut new_manager_records = before.manager_records.clone();
111 if let Some(existing) = new_manager_records
112 .iter_mut()
113 .find(|record| record.id == thread.id)
114 {
115 *existing = thread.clone();
116 } else {
117 new_manager_records.push(thread.clone());
118 }
119 let encode_records = |records: &[Thread]| -> Result<Vec<Vec<u8>>> {
120 records
121 .iter()
122 .map(|record| {
123 manager
124 .encode_thread_record_snapshot(record)
125 .map_err(Into::into)
126 })
127 .collect()
128 };
129 let snapshots = ThreadUpdateSnapshots::from_record_sets(
130 before.manager_snapshot,
131 new_manager_snapshot,
132 encode_records(&before.manager_records)?,
133 encode_records(&new_manager_records)?,
134 before.ref_absent,
135 );
136 let thread_name = ThreadName::new(&thread.thread);
137 if repo.refs().get_thread(&thread_name)? == Some(new_state) {
138 objects::fault_inject::maybe_fail_at("thread_manager_save_in_thread_update")?;
139 manager.save(thread)?;
140 repo.oplog().record_thread_update(
141 &thread_name,
142 &before.state,
143 &new_state,
144 snapshots,
145 Some(&repo.op_scope()),
146 )?;
147 } else {
148 objects::fault_inject::maybe_fail_at("thread_manager_save_in_thread_update")?;
149 repo.commit_and_publish(
150 vec![OpRecord::ThreadUpdate {
151 name: thread_name.to_string(),
152 old_state: before.state,
153 new_state,
154 manager_snapshots: snapshots,
155 }],
156 &[RefUpdate::Thread {
157 name: thread_name,
158 expected: if before.ref_absent {
159 RefExpectation::Missing
160 } else {
161 RefExpectation::Value(before.state)
162 },
163 new: Some(new_state),
164 }],
165 )?;
166 manager.save(thread)?;
167 }
168 Ok(())
169}
170
171impl ThreadListOptions {
172 pub fn new() -> Self {
173 Self::default()
174 }
175
176 pub fn include_auto(mut self, include_auto: bool) -> Self {
177 self.include_auto = include_auto;
178 self
179 }
180
181 pub fn include_abandoned(mut self, include_abandoned: bool) -> Self {
182 self.include_abandoned = include_abandoned;
183 self
184 }
185}
186
187#[derive(Debug, Clone, Serialize)]
193pub struct ThreadListReport {
194 pub output_kind: &'static str,
195 pub threads: Vec<ThreadListEntry>,
196 pub available_git_refs: Vec<AvailableGitRef>,
197 pub current: Option<String>,
198}
199
200#[derive(Debug, Clone, Serialize, JsonSchema)]
204pub struct ThreadListEntry {
205 pub name: String,
206 pub operation: Option<RepositoryOperationStatus>,
207 pub remote_tracking: Option<GitRemoteTrackingStatus>,
208 pub base_state: Option<String>,
209 pub base_root: Option<String>,
210 pub current_state: Option<String>,
211 pub path: Option<String>,
212 pub execution_path: Option<String>,
213 #[serde(skip_serializing_if = "Option::is_none")]
214 pub session_id: Option<String>,
215 #[serde(skip_serializing_if = "Option::is_none")]
216 pub heddle_session_id: Option<String>,
217 pub actor: Option<ThreadActorInfo>,
218 pub harness: Option<String>,
219 pub thinking_level: Option<String>,
220 #[serde(skip_serializing_if = "Option::is_none")]
221 pub native_actor_key: Option<String>,
222 #[serde(skip_serializing_if = "Option::is_none")]
223 pub native_parent_actor_key: Option<String>,
224 #[serde(skip_serializing_if = "Option::is_none")]
225 pub probe_source: Option<String>,
226 #[serde(skip_serializing_if = "Option::is_none")]
227 pub probe_confidence: Option<f32>,
228 pub usage_summary: Option<AgentUsageSummary>,
229 pub last_progress_at: Option<String>,
230 pub last_activity_at: Option<String>,
231 pub report_flush_state: Option<String>,
232 pub attach_reason: Option<String>,
233 pub thread_mode: Option<ThreadMode>,
234 pub thread_state: Option<ThreadState>,
235 pub freshness: Option<ThreadFreshness>,
236 pub visibility: String,
237 pub target_thread: Option<String>,
238 pub parent_thread: Option<String>,
239 pub child_threads: Vec<String>,
240 pub sibling_threads: Vec<String>,
241 pub stack_depth: usize,
242 pub stale_from_parent: bool,
243 pub task: Option<String>,
244 pub task_assignment_id: Option<String>,
245 pub task_summary: Option<ThreadTaskSummary>,
246 pub changed_paths: Vec<String>,
247 pub promotion_suggested: bool,
248 pub impact_categories: Vec<ThreadImpactCategory>,
249 pub heavy_impact_paths: Vec<String>,
250 pub verification_summary: ThreadVerificationSummary,
251 pub confidence_summary: ThreadConfidenceSummary,
252 pub integration_policy_result: ThreadIntegrationPolicy,
253 pub coordination_status: CoordinationStatus,
254 pub is_current: bool,
255 pub is_isolated: bool,
256 pub thread_health: String,
257 pub blockers: Vec<String>,
258 #[serde(serialize_with = "serialize_empty_action_as_null")]
259 #[schemars(with = "Option<String>")]
260 pub recommended_action: String,
261 pub recommended_action_template: Option<ActionTemplate>,
262 pub git_branch_tip: Option<String>,
263 pub history_imported: bool,
264 pub auto: bool,
269 pub shared_target_dir: Option<String>,
271}
272
273pub type ThreadSummary = ThreadListEntry;
275
276#[derive(Debug, Clone, Serialize, JsonSchema)]
278pub struct AvailableGitRef {
279 pub name: String,
280 pub git_commit: String,
281 #[serde(serialize_with = "serialize_empty_action_as_null")]
282 #[schemars(with = "Option<String>")]
283 pub recommended_action: String,
284 pub recommended_action_template: Option<ActionTemplate>,
285}
286
287#[derive(Debug, Clone, Serialize, PartialEq, Eq, JsonSchema)]
289pub struct ThreadActorInfo {
290 #[serde(skip_serializing_if = "Option::is_none")]
291 pub provider: Option<String>,
292 #[serde(skip_serializing_if = "Option::is_none")]
293 pub model: Option<String>,
294}
295
296#[derive(Debug, Clone, Serialize, PartialEq, Eq, JsonSchema)]
298pub struct ThreadTaskSummary {
299 pub task_id: String,
300 pub title: String,
301 pub status: String,
302 pub target_thread: String,
303 pub updated_at: String,
304 pub completed_at: Option<String>,
305 pub coordination_discussion_id: Option<String>,
306}
307
308impl From<&AgentTaskRecord> for ThreadTaskSummary {
309 fn from(task: &AgentTaskRecord) -> Self {
310 Self {
311 task_id: task.task_id.clone(),
312 title: task.title.clone(),
313 status: task.status.to_string(),
314 target_thread: task.target_thread.clone(),
315 updated_at: task.updated_at.to_rfc3339(),
316 completed_at: task.completed_at.map(|time| time.to_rfc3339()),
317 coordination_discussion_id: task.coordination_discussion_id.clone(),
318 }
319 }
320}
321
322impl ThreadListEntry {
323 fn from_view(view: ThreadView, coordination_status: CoordinationStatus) -> Self {
324 let mode = view.record.mode.clone();
325 Self {
326 name: view.record.thread,
327 operation: None,
328 remote_tracking: None,
329 base_state: Some(view.record.base_state),
330 base_root: Some(view.record.base_root),
331 current_state: view.record.current_state,
332 path: view
333 .runtime
334 .materialized_path
335 .as_ref()
336 .or(view.runtime.path.as_ref())
337 .and_then(|p| display_path_string(p)),
338 execution_path: view
339 .runtime
340 .execution_path
341 .as_ref()
342 .and_then(|p| display_path_string(p)),
343 session_id: view.runtime.session_id,
344 heddle_session_id: view.runtime.heddle_session_id,
345 actor: match (view.runtime.provider, view.runtime.model) {
346 (None, None) => None,
347 (provider, model) => Some(ThreadActorInfo { provider, model }),
348 },
349 harness: view.runtime.harness,
350 thinking_level: view.runtime.thinking_level,
351 native_actor_key: view.runtime.native_actor_key,
352 native_parent_actor_key: view.runtime.native_parent_actor_key,
353 probe_source: view.runtime.probe_source,
354 probe_confidence: view.runtime.probe_confidence,
355 usage_summary: view.runtime.usage_summary,
356 last_progress_at: view.runtime.last_progress_at.map(|ts| ts.to_rfc3339()),
357 last_activity_at: Some(view.record.updated_at.to_rfc3339()),
358 report_flush_state: view.runtime.report_flush_state,
359 attach_reason: view.runtime.attach_reason,
360 thread_mode: Some(mode.clone()),
361 thread_state: Some(view.record.state),
362 freshness: Some(view.record.freshness),
363 visibility: if view.is_isolated {
364 visibility_label(&mode).to_string()
365 } else {
366 "ref_only".to_string()
367 },
368 target_thread: view.record.target_thread,
369 parent_thread: view.record.parent_thread,
370 child_threads: Vec::new(),
371 sibling_threads: Vec::new(),
372 stack_depth: 0,
373 stale_from_parent: false,
374 task: view.record.task,
375 task_assignment_id: None,
376 task_summary: None,
377 changed_paths: view.record.changed_paths,
378 promotion_suggested: view.record.promotion_suggested,
379 impact_categories: view.record.impact_categories,
380 heavy_impact_paths: view.record.heavy_impact_paths,
381 verification_summary: view.record.verification_summary,
382 confidence_summary: view.record.confidence_summary,
383 integration_policy_result: view.record.integration_policy_result,
384 coordination_status,
385 is_current: view.is_current,
386 is_isolated: view.is_isolated,
387 thread_health: "clean".to_string(),
388 blockers: Vec::new(),
389 recommended_action: String::new(),
390 recommended_action_template: None,
391 git_branch_tip: None,
392 history_imported: true,
393 auto: view.record.auto,
394 shared_target_dir: view
395 .record
396 .shared_target_dir
397 .as_ref()
398 .map(|p| p.display().to_string()),
399 }
400 }
401}
402
403fn display_path_string(path: &Path) -> Option<String> {
404 let rendered = path.display().to_string();
405 if rendered.trim().is_empty() {
406 None
407 } else {
408 Some(rendered)
409 }
410}
411
412pub fn list_threads(repo: &Repository, options: ThreadListOptions) -> Result<ThreadListReport> {
414 list_threads_inner(repo, options, None)
415}
416
417pub fn list_threads_with_worktree_status(
420 repo: &Repository,
421 options: ThreadListOptions,
422 worktree_status: &WorktreeStatus,
423) -> Result<ThreadListReport> {
424 list_threads_inner(repo, options, Some(worktree_status))
425}
426
427fn list_threads_inner(
428 repo: &Repository,
429 options: ThreadListOptions,
430 worktree_status: Option<&WorktreeStatus>,
431) -> Result<ThreadListReport> {
432 let mut summaries = collect_thread_summaries_inner(repo, worktree_status)?;
433 if !options.include_auto {
434 summaries.retain(|summary| summary.is_current || !summary.auto);
438 }
439 if !options.include_abandoned {
440 summaries.retain(|summary| {
441 summary.is_current || !matches!(summary.thread_state, Some(ThreadState::Abandoned))
442 });
443 }
444 let available_git_refs = split_available_git_refs(&mut summaries);
445 let current = summaries
446 .iter()
447 .find(|summary| summary.is_current)
448 .map(|summary| summary.name.clone())
449 .or(repo.current_lane()?);
450 Ok(ThreadListReport {
451 output_kind: "thread_list",
452 threads: summaries,
453 available_git_refs,
454 current,
455 })
456}
457
458pub fn collect_thread_summaries(repo: &Repository) -> Result<Vec<ThreadListEntry>> {
460 collect_thread_summaries_inner(repo, None)
461}
462
463fn collect_thread_summaries_inner(
464 repo: &Repository,
465 worktree_status: Option<&WorktreeStatus>,
466) -> Result<Vec<ThreadListEntry>> {
467 let thread_refs = repo.refs().list_threads_with_states()?;
468 let thread_ref_names = thread_refs
469 .iter()
470 .map(|(name, _)| name.to_string())
471 .collect::<HashSet<_>>();
472 let thread_ref_states = thread_refs
473 .iter()
474 .map(|(name, state)| (name.to_string(), *state))
475 .collect::<HashMap<_, _>>();
476 let current = repo.current_lane()?;
477 let operation = repo.operation_status()?;
478 let remote_tracking = repo.git_remote_tracking_status().unwrap_or(None);
479 let import_hint = repo.git_import_guidance().unwrap_or(None);
480 let branch_tips = repo
481 .git_overlay_branch_tips()
482 .unwrap_or_default()
483 .into_iter()
484 .map(|tip| (tip.branch.clone(), tip))
485 .collect::<HashMap<_, _>>();
486 let registry = ActorPresenceStore::new(repo.heddle_dir());
487 let task_store = AgentTaskStore::new(repo.heddle_dir());
488 let thread_manager = ThreadManager::new(repo.heddle_dir());
489 let mut entries_by_thread: HashMap<String, Vec<ActorPresence>> = HashMap::new();
490 let mut threads_by_name: HashMap<String, Thread> = HashMap::new();
491 for entry in registry.list()? {
492 entries_by_thread
493 .entry(entry.thread.clone())
494 .or_default()
495 .push(entry);
496 }
497 for mut thread in thread_manager.list()? {
498 refresh_thread_freshness(repo, &mut thread)?;
499 threads_by_name.insert(thread.thread.clone(), thread);
500 }
501
502 let mut names: BTreeSet<String> = thread_ref_names.iter().cloned().collect();
503 names.extend(current.iter().cloned());
504 names.extend(entries_by_thread.keys().cloned());
505 names.extend(threads_by_name.keys().cloned());
506 names.extend(branch_tips.keys().cloned());
507
508 let mut summaries = Vec::new();
509 for name in names {
510 let entries = entries_by_thread.remove(&name).unwrap_or_default();
511 let task_assignment_id = task_assignment_id_from_entries(&entries);
512 let task_summary = task_summary_for_assignment(&task_store, task_assignment_id.as_deref())?;
513 let (view, coordination_status) = build_thread_view(
514 repo,
515 current.as_ref() == Some(&name),
516 name.clone(),
517 thread_ref_states.get(&name).copied(),
518 entries,
519 threads_by_name.remove(&name),
520 branch_tips.get(&name).cloned(),
521 )?;
522 let mut summary = ThreadListEntry::from_view(view, coordination_status);
523 summary.task_assignment_id = task_assignment_id;
524 summary.task_summary = task_summary;
525 if let Some(branch_tip) = branch_tips.get(&summary.name) {
526 summary.git_branch_tip = Some(branch_tip.git_commit.clone());
527 summary.history_imported = branch_tip.history_imported;
528 }
529 let has_heddle_tip = thread_ref_names.contains(&summary.name);
530 let needs_advice = summary.thread_state != Some(ThreadState::Active)
531 || summary.freshness == Some(ThreadFreshness::Stale)
532 || !summary.changed_paths.is_empty()
533 || summary.promotion_suggested;
534 if needs_advice {
535 let thread = Thread {
536 id: summary.name.clone(),
537 thread: summary.name.clone(),
538 target_thread: summary.target_thread.clone(),
539 parent_thread: summary.parent_thread.clone(),
540 mode: summary
541 .thread_mode
542 .clone()
543 .unwrap_or(ThreadMode::Materialized),
544 state: summary.thread_state.clone().unwrap_or(ThreadState::Active),
545 base_state: summary.base_state.clone().unwrap_or_default(),
546 base_root: summary.base_root.clone().unwrap_or_default(),
547 current_state: summary.current_state.clone(),
548 merged_state: None,
549 task: summary.task.clone(),
550 execution_path: summary
551 .execution_path
552 .as_ref()
553 .map(PathBuf::from)
554 .unwrap_or_else(|| repo.root().to_path_buf()),
555 materialized_path: summary.path.as_ref().map(PathBuf::from),
556 changed_paths: summary.changed_paths.clone(),
557 impact_categories: summary.impact_categories.clone(),
558 heavy_impact_paths: summary.heavy_impact_paths.clone(),
559 promotion_suggested: summary.promotion_suggested,
560 freshness: summary
561 .freshness
562 .clone()
563 .unwrap_or(ThreadFreshness::Unknown),
564 verification_summary: summary.verification_summary.clone(),
565 confidence_summary: summary.confidence_summary.clone(),
566 integration_policy_result: summary.integration_policy_result.clone(),
567 created_at: Utc::now(),
568 updated_at: Utc::now(),
569 ephemeral: None,
570 auto: summary.auto,
571 shared_target_dir: summary.shared_target_dir.as_ref().map(PathBuf::from),
572 };
573 let advice = describe_thread_advice(&thread, false, 0, false);
574 summary.thread_health = advice.thread_health;
575 summary.blockers = advice.blockers;
576 summary.recommended_action = advice.recommended_action;
577 }
578 apply_terminal_thread_advice(&mut summary);
579 apply_materialized_merge_advice(repo, &mut summary);
580 if let Some(branch_tip) = branch_tips.get(&summary.name)
581 && !has_heddle_tip
582 {
583 summary.blockers.clear();
584 summary.thread_health = if branch_tip.history_imported {
585 "imported".to_string()
586 } else {
587 "git_backed".to_string()
588 };
589 if summary.is_current {
590 summary.recommended_action.clear();
591 } else {
592 summary.recommended_action = if branch_tip.branch.starts_with('-') {
593 format!(
594 "heddle thread switch -- {}",
595 shell_quote(&branch_tip.branch)
596 )
597 } else {
598 format!("heddle thread switch {}", shell_quote(&branch_tip.branch))
599 };
600 }
601 }
602 if repo.capability() == repo::RepositoryCapability::GitOverlay
603 && summary.history_imported
604 && summary.current_state.is_some()
605 && remote_tracking_local_ref(repo, &summary.name).is_some()
606 {
607 summary.thread_health = "remote_tracking".to_string();
608 summary.coordination_status = CoordinationStatus::Clean;
609 summary.blockers.clear();
610 summary.recommended_action =
611 canonical_git_repair_ref_preview_command(None, &summary.name);
612 }
613 if summary.is_current {
614 enrich_current_summary_with_dirty_paths(repo, &mut summary, worktree_status)?;
615 summary.operation = operation.clone();
616 summary.remote_tracking = remote_tracking.clone();
617 summary.recommended_action = effective_next_action(
618 NextActionInput::default(
619 operation.as_ref(),
620 remote_tracking.as_ref(),
621 import_hint.as_ref(),
622 Some(&summary.recommended_action),
623 )
624 .with_source_authority(repo.source_authority())
625 .current_thread(Some(&summary.thread_health)),
626 );
627 summary.recommended_action = contextual_thread_action(
628 repo,
629 &summary.name,
630 summary.target_thread.as_deref(),
631 &summary.recommended_action,
632 );
633 }
634 summaries.push(summary);
635 }
636
637 let mut children_by_parent: HashMap<String, Vec<String>> = HashMap::new();
638 for summary in &summaries {
639 if let Some(parent) = &summary.parent_thread {
640 children_by_parent
641 .entry(parent.clone())
642 .or_default()
643 .push(summary.name.clone());
644 }
645 }
646 for summary in &mut summaries {
647 if let Some(children) = children_by_parent.get(&summary.name) {
648 let mut children = children.clone();
649 children.sort();
650 summary.child_threads = children;
651 }
652 }
653
654 let parents_by_name = summaries
655 .iter()
656 .map(|summary| (summary.name.clone(), summary.parent_thread.clone()))
657 .collect::<HashMap<_, _>>();
658 for summary in &mut summaries {
659 summary.sibling_threads = summary
660 .parent_thread
661 .as_ref()
662 .and_then(|parent| children_by_parent.get(parent))
663 .map(|children| {
664 children
665 .iter()
666 .filter(|child| *child != &summary.name)
667 .cloned()
668 .collect()
669 })
670 .unwrap_or_default();
671 summary.stack_depth = stack_depth(&parents_by_name, &summary.name);
672 summary.stale_from_parent =
673 summary.parent_thread.is_some() && summary.freshness == Some(ThreadFreshness::Stale);
674 if summary.last_progress_at.is_some() {
675 summary.last_activity_at = summary.last_progress_at.clone();
676 }
677 summary.recommended_action_template = action_template(&summary.recommended_action);
678 }
679
680 summaries.sort_by(|a, b| a.name.cmp(&b.name));
681 Ok(summaries)
682}
683
684pub fn find_thread_summary(repo: &Repository, name: &str) -> Result<Option<ThreadListEntry>> {
686 Ok(collect_thread_summaries(repo)?
687 .into_iter()
688 .find(|summary| summary.name == name))
689}
690
691pub fn split_available_git_refs(summaries: &mut Vec<ThreadListEntry>) -> Vec<AvailableGitRef> {
693 let mut available = Vec::new();
694 summaries.retain(|summary| {
695 if thread_is_available_git_ref(summary) {
696 available.push(available_git_ref_from_summary(summary));
697 false
698 } else {
699 true
700 }
701 });
702 available
703}
704
705pub fn thread_is_imported_git_ref(entry: &ThreadListEntry) -> bool {
707 !entry.is_current
708 && entry.path.is_none()
709 && entry.execution_path.is_none()
710 && entry.target_thread.is_none()
711 && entry.current_state.is_some()
712 && entry.history_imported
713 && (entry.git_branch_tip.is_some() || entry.name.starts_with("origin/"))
714}
715
716pub fn thread_is_available_git_ref(entry: &ThreadListEntry) -> bool {
718 !entry.is_current
719 && entry.path.is_none()
720 && entry.execution_path.is_none()
721 && entry.target_thread.is_none()
722 && entry.current_state.is_none()
723 && entry.git_branch_tip.is_some()
724}
725
726pub fn visibility_label(mode: &ThreadMode) -> &'static str {
728 match mode {
729 ThreadMode::Materialized => "materialized",
730 ThreadMode::Virtualized => "virtualized",
731 ThreadMode::Solid => "solid",
732 }
733}
734
735fn available_git_ref_from_summary(summary: &ThreadListEntry) -> AvailableGitRef {
736 AvailableGitRef {
737 name: summary.name.clone(),
738 git_commit: summary.git_branch_tip.clone().unwrap_or_default(),
739 recommended_action: summary.recommended_action.clone(),
740 recommended_action_template: summary
741 .recommended_action_template
742 .clone()
743 .or_else(|| action_template(&summary.recommended_action)),
744 }
745}
746
747fn enrich_current_summary_with_dirty_paths(
748 repo: &Repository,
749 summary: &mut ThreadListEntry,
750 worktree_status: Option<&WorktreeStatus>,
751) -> Result<()> {
752 let status = match worktree_status {
753 Some(status) => status.clone(),
754 None => {
755 let baseline = match repo.current_state_for_worktree_status()? {
756 Some(state) => repo.require_tree_for_worktree_status(&state.tree)?,
757 None => Tree::new(),
758 };
759 let options = repo::resolve_worktree_status_options(None, Some(repo.config()));
760 repo.compare_worktree_cached_with_options(&baseline, &options)?
761 }
762 };
763 let mut paths = summary
764 .changed_paths
765 .iter()
766 .cloned()
767 .collect::<BTreeSet<_>>();
768 paths.extend(
769 status
770 .modified
771 .iter()
772 .chain(status.added.iter())
773 .chain(status.deleted.iter())
774 .map(|path| path.to_string_lossy().to_string()),
775 );
776 summary.changed_paths = paths.into_iter().collect();
777 Ok(())
778}
779
780fn stack_depth(parents_by_name: &HashMap<String, Option<String>>, thread: &str) -> usize {
781 let mut depth = 0usize;
782 let mut cursor = parents_by_name.get(thread).cloned().flatten();
783 while let Some(parent) = cursor {
784 depth += 1;
785 cursor = parents_by_name.get(&parent).cloned().flatten();
786 }
787 depth
788}
789
790fn primary_agent_entry(entries: &[ActorPresence]) -> Option<&ActorPresence> {
791 entries
792 .iter()
793 .filter(|entry| entry.status == ActorPresenceStatus::Active)
794 .max_by_key(|entry| entry.started_at)
795 .or_else(|| entries.iter().max_by_key(|entry| entry.started_at))
796}
797
798fn task_assignment_id_from_entries(entries: &[ActorPresence]) -> Option<String> {
799 primary_agent_entry(entries).and_then(|entry| entry.task_assignment_id.clone())
800}
801
802fn task_summary_for_assignment(
803 store: &AgentTaskStore,
804 task_assignment_id: Option<&str>,
805) -> Result<Option<ThreadTaskSummary>> {
806 let Some(task_assignment_id) = task_assignment_id else {
807 return Ok(None);
808 };
809 Ok(store
810 .load(task_assignment_id)?
811 .as_ref()
812 .map(ThreadTaskSummary::from))
813}
814
815fn build_thread_view(
816 repo: &Repository,
817 is_current: bool,
818 name: String,
819 ref_state: Option<objects::object::StateId>,
820 entries: Vec<ActorPresence>,
821 thread: Option<Thread>,
822 branch_tip: Option<GitOverlayBranchTip>,
823) -> Result<(ThreadView, CoordinationStatus)> {
824 let current_state = ref_state
825 .or_else(|| {
826 (is_current && repo.capability() == repo::RepositoryCapability::GitOverlay)
827 .then(|| {
828 branch_tip
829 .as_ref()
830 .and_then(|tip| tip.mapped_state)
831 .or_else(|| {
832 repo.git_overlay_mapped_state_for_branch(&name)
833 .ok()
834 .flatten()
835 })
836 })
837 .flatten()
838 })
839 .map(|id| id.short());
840 let has_heddle_tip = current_state.is_some();
841 let active: Vec<&ActorPresence> = entries
842 .iter()
843 .filter(|entry| entry.status == ActorPresenceStatus::Active)
844 .collect();
845 let complete: Vec<&ActorPresence> = entries
846 .iter()
847 .filter(|entry| entry.status == ActorPresenceStatus::Complete)
848 .collect();
849
850 let primary = active
851 .iter()
852 .max_by_key(|entry| entry.started_at)
853 .copied()
854 .or_else(|| entries.iter().max_by_key(|entry| entry.started_at));
855 let base_state = thread
856 .as_ref()
857 .map(|thread| thread.base_state.clone())
858 .or_else(|| primary.map(|entry| entry.base_state.clone()))
859 .or(current_state.clone());
860 let base_root = thread.as_ref().map(|thread| thread.base_root.clone());
861 let runtime = ThreadRuntimeOverlay {
862 path: thread
863 .as_ref()
864 .and_then(|thread| thread.materialized_path.clone())
865 .or_else(|| primary.and_then(|entry| entry.path.clone())),
866 execution_path: thread.as_ref().map(|thread| thread.execution_path.clone()),
867 materialized_path: thread
868 .as_ref()
869 .and_then(|thread| thread.materialized_path.clone()),
870 session_id: primary.map(|entry| entry.session_id.clone()),
871 heddle_session_id: primary.and_then(|entry| entry.heddle_session_id.clone()),
872 harness: primary.and_then(|entry| entry.harness.clone()),
873 thinking_level: primary.and_then(|entry| entry.thinking_level.clone()),
874 native_actor_key: primary.and_then(|entry| entry.native_actor_key.clone()),
875 native_parent_actor_key: primary.and_then(|entry| entry.native_parent_actor_key.clone()),
876 probe_source: primary.and_then(|entry| entry.probe_source.clone()),
877 probe_confidence: primary.and_then(|entry| entry.probe_confidence),
878 usage_summary: primary.map(|entry| entry.usage_summary.clone()),
879 last_progress_at: primary.and_then(|entry| entry.last_progress_at),
880 report_flush_state: primary.and_then(|entry| entry.report_flush_state.clone()),
881 attach_reason: primary.and_then(|entry| entry.attach_reason.clone()),
882 provider: primary.and_then(|entry| entry.provider.clone()),
883 model: primary.and_then(|entry| entry.model.clone()),
884 thread_mode: thread.as_ref().map(|thread| thread.mode.clone()),
885 thread_state: thread.as_ref().map(|thread| thread.state.clone()),
886 };
887 let thread_record = thread.as_ref().map(|thread| thread.to_record());
888 let thread_state_for_status = thread_record.as_ref().map(|thread| thread.state.clone());
889 let coordination_status = if matches!(
890 thread_state_for_status,
891 Some(ThreadState::Merged | ThreadState::Abandoned)
892 ) {
893 CoordinationStatus::Clean
894 } else if thread_state_for_status == Some(ThreadState::Blocked) {
895 CoordinationStatus::Blocked
896 } else if thread_state_for_status == Some(ThreadState::Ready) {
897 CoordinationStatus::MergeReady
898 } else if active.len() > 1 {
899 CoordinationStatus::Blocked
900 } else if !active.is_empty()
901 && complete
902 .iter()
903 .any(|entry| entry.base_state != active[0].base_state)
904 {
905 CoordinationStatus::Diverged
906 } else if !complete.is_empty() {
907 CoordinationStatus::MergeReady
908 } else if base_state.is_some() && current_state.is_some() && base_state != current_state {
909 CoordinationStatus::Ahead
910 } else {
911 CoordinationStatus::Clean
912 };
913
914 let view = match thread {
915 Some(mut thread) => {
916 thread.current_state = current_state;
917 thread.to_view(runtime, is_current)
918 }
919 None => ThreadView::from_record(
920 repo::ThreadRecord {
921 id: name.clone(),
922 thread: name.clone(),
923 target_thread: None,
924 parent_thread: None,
925 mode: ThreadMode::Materialized,
926 state: ThreadState::Active,
927 base_state: base_state.unwrap_or_default(),
928 base_root: base_root.unwrap_or_default(),
929 current_state,
930 merged_state: None,
931 task: None,
932 changed_paths: Vec::new(),
933 impact_categories: Vec::new(),
934 heavy_impact_paths: Vec::new(),
935 promotion_suggested: false,
936 freshness: ThreadFreshness::Unknown,
937 verification_summary: Default::default(),
938 confidence_summary: Default::default(),
939 integration_policy_result: Default::default(),
940 created_at: Utc::now(),
941 updated_at: Utc::now(),
942 ephemeral: None,
943 auto: false,
944 shared_target_dir: None,
945 },
946 runtime,
947 is_current,
948 ),
949 };
950
951 if let Some(branch_tip) = branch_tip
952 && !has_heddle_tip
953 && view.record.current_state.is_none()
954 {
955 let mut record = view.record.clone();
956 record.current_state = None;
957 let mut runtime = view.runtime.clone();
958 if runtime.attach_reason.is_none() {
959 runtime.attach_reason = Some(format!(
960 "using Git-backed branch tip {}",
961 branch_tip.git_commit
962 ));
963 }
964 return Ok((
965 ThreadView::from_record(record, runtime, is_current),
966 coordination_status,
967 ));
968 }
969
970 Ok((view, coordination_status))
971}
972
973fn apply_materialized_merge_advice(repo: &Repository, summary: &mut ThreadListEntry) {
974 let Some(action) = materialized_merge_resolve_action(repo, summary) else {
975 return;
976 };
977 summary.thread_health = "blocked".to_string();
978 if summary.blockers.is_empty() {
979 summary
980 .blockers
981 .push("Merge conflicts need resolution".to_string());
982 }
983 summary.recommended_action = action;
984 summary.recommended_action_template = action_template(&summary.recommended_action);
985}
986
987fn materialized_merge_resolve_action(
988 repo: &Repository,
989 summary: &ThreadListEntry,
990) -> Option<String> {
991 if let Some(path) = summary.execution_path.as_deref() {
992 let path = PathBuf::from(path);
993 if !path.exists() {
994 return None;
995 }
996 let thread_repo = Repository::open(&path).ok()?;
997 return thread_repo
998 .merge_state_manager()
999 .is_merge_in_progress()
1000 .then(|| {
1001 heddle_action(vec![
1002 "--repo".to_string(),
1003 path.display().to_string(),
1004 "resolve".to_string(),
1005 "--list".to_string(),
1006 ])
1007 });
1008 }
1009
1010 (summary.is_current && repo.merge_state_manager().is_merge_in_progress())
1011 .then(|| heddle_action(["resolve", "--list"]))
1012}
1013
1014fn apply_terminal_thread_advice(summary: &mut ThreadListEntry) {
1015 match summary.thread_state {
1016 Some(ThreadState::Merged) => {
1017 summary.thread_health = "clean".to_string();
1018 summary.blockers.clear();
1019 summary.recommended_action = "heddle thread cleanup --merged --dry-run".to_string();
1020 summary.coordination_status = CoordinationStatus::Clean;
1021 }
1022 Some(ThreadState::Abandoned) => {
1023 summary.thread_health = "clean".to_string();
1024 summary.blockers.clear();
1025 summary.recommended_action.clear();
1026 summary.coordination_status = CoordinationStatus::Clean;
1027 }
1028 _ => {}
1029 }
1030}
1031
1032fn remote_tracking_local_ref(repo: &Repository, thread_name: &str) -> Option<String> {
1033 let git = SleyRepository::discover(repo.root()).ok()?;
1034 let remotes = git.remote_names().ok()?;
1035 remotes
1036 .iter()
1037 .find_map(|remote| thread_name.strip_prefix(&format!("{remote}/")))
1038 .filter(|branch| !branch.is_empty())
1039 .map(str::to_string)
1040}
1041
1042#[cfg(test)]
1043mod tests {
1044 use chrono::Utc;
1045 use objects::object::ThreadName;
1046 use repo::{
1047 Thread, ThreadConfidenceSummary, ThreadFreshness, ThreadIntegrationPolicy, ThreadManager,
1048 ThreadMode, ThreadState, ThreadVerificationSummary,
1049 };
1050 use tempfile::TempDir;
1051
1052 use super::*;
1053
1054 fn sample_thread(name: &str, auto: bool) -> Thread {
1055 Thread {
1056 id: name.to_string(),
1057 thread: name.to_string(),
1058 target_thread: None,
1059 parent_thread: None,
1060 mode: ThreadMode::Materialized,
1061 state: ThreadState::Active,
1062 base_state: String::new(),
1063 base_root: String::new(),
1064 current_state: None,
1065 merged_state: None,
1066 task: None,
1067 execution_path: PathBuf::from("/tmp"),
1068 materialized_path: None,
1069 changed_paths: Vec::new(),
1070 impact_categories: Vec::new(),
1071 heavy_impact_paths: Vec::new(),
1072 promotion_suggested: false,
1073 freshness: ThreadFreshness::Unknown,
1074 verification_summary: ThreadVerificationSummary::default(),
1075 confidence_summary: ThreadConfidenceSummary::default(),
1076 integration_policy_result: ThreadIntegrationPolicy::default(),
1077 created_at: Utc::now(),
1078 updated_at: Utc::now(),
1079 ephemeral: None,
1080 auto,
1081 shared_target_dir: None,
1082 }
1083 }
1084
1085 #[test]
1086 fn list_threads_empty_repo_returns_empty_domain_lists() {
1087 let temp = TempDir::new().unwrap();
1088 let repo = Repository::init_default(temp.path()).unwrap();
1089 let report = list_threads(&repo, ThreadListOptions::new()).unwrap();
1092 assert_eq!(report.output_kind, "thread_list");
1093 assert!(report.available_git_refs.is_empty());
1094 assert!(
1096 report
1097 .threads
1098 .iter()
1099 .all(|t| t.name == "main" || t.is_current),
1100 "unexpected threads: {:?}",
1101 report.threads.iter().map(|t| &t.name).collect::<Vec<_>>()
1102 );
1103 let value = serde_json::to_value(&report).unwrap();
1104 assert_eq!(value["output_kind"], "thread_list");
1105 assert!(value["threads"].is_array());
1106 assert!(value["available_git_refs"].is_array());
1107 }
1108
1109 #[test]
1110 fn list_threads_sorts_by_name() {
1111 let temp = TempDir::new().unwrap();
1112 let repo = Repository::init_default(temp.path()).unwrap();
1113 let manager = ThreadManager::new(repo.heddle_dir());
1114 for name in ["zeta", "alpha", "mid"] {
1115 manager.save(&sample_thread(name, false)).unwrap();
1116 let _ = repo.refs().set_thread(
1117 &ThreadName::new(name),
1118 &objects::object::StateId::from_bytes([0u8; 32]),
1119 );
1120 }
1121
1122 let report = list_threads(&repo, ThreadListOptions::new().include_auto(true)).unwrap();
1123 let names: Vec<&str> = report.threads.iter().map(|t| t.name.as_str()).collect();
1124 let mut sorted = names.clone();
1125 sorted.sort();
1126 assert_eq!(names, sorted, "thread list must be sorted by name");
1127 assert!(names.contains(&"alpha"));
1128 assert!(names.contains(&"mid"));
1129 assert!(names.contains(&"zeta"));
1130 }
1131
1132 #[test]
1133 fn list_threads_hides_auto_unless_include_auto_or_current() {
1134 let temp = TempDir::new().unwrap();
1135 let repo = Repository::init_default(temp.path()).unwrap();
1136 let manager = ThreadManager::new(repo.heddle_dir());
1137 manager.save(&sample_thread("user-feature", false)).unwrap();
1138 manager
1139 .save(&sample_thread("harness-session", true))
1140 .unwrap();
1141
1142 let filtered = list_threads(&repo, ThreadListOptions::new()).unwrap();
1143 let filtered_names: Vec<&str> = filtered.threads.iter().map(|t| t.name.as_str()).collect();
1144 assert!(
1145 filtered_names.contains(&"user-feature"),
1146 "user thread should remain: {filtered_names:?}"
1147 );
1148 assert!(
1149 !filtered_names.contains(&"harness-session"),
1150 "auto thread should be hidden by default: {filtered_names:?}"
1151 );
1152
1153 let all = list_threads(&repo, ThreadListOptions::new().include_auto(true)).unwrap();
1154 let all_names: Vec<&str> = all.threads.iter().map(|t| t.name.as_str()).collect();
1155 assert!(
1156 all_names.contains(&"harness-session"),
1157 "include_auto must surface harness thread: {all_names:?}"
1158 );
1159 }
1160
1161 #[test]
1162 fn available_git_ref_serializes_empty_recommended_action_as_null() {
1163 let value = serde_json::to_value(AvailableGitRef {
1164 name: "main".to_string(),
1165 git_commit: "0123456789abcdef".to_string(),
1166 recommended_action: String::new(),
1167 recommended_action_template: None,
1168 })
1169 .unwrap();
1170 assert!(value["recommended_action"].is_null());
1171 }
1172
1173 #[test]
1174 fn split_available_git_refs_moves_git_only_tips() {
1175 let mut summaries = vec![
1176 ThreadListEntry {
1177 name: "feature".into(),
1178 operation: None,
1179 remote_tracking: None,
1180 base_state: None,
1181 base_root: None,
1182 current_state: Some("abc".into()),
1183 path: None,
1184 execution_path: None,
1185 session_id: None,
1186 heddle_session_id: None,
1187 actor: None,
1188 harness: None,
1189 thinking_level: None,
1190 native_actor_key: None,
1191 native_parent_actor_key: None,
1192 probe_source: None,
1193 probe_confidence: None,
1194 usage_summary: None,
1195 last_progress_at: None,
1196 last_activity_at: None,
1197 report_flush_state: None,
1198 attach_reason: None,
1199 thread_mode: None,
1200 thread_state: None,
1201 freshness: None,
1202 visibility: "ref_only".into(),
1203 target_thread: None,
1204 parent_thread: None,
1205 child_threads: vec![],
1206 sibling_threads: vec![],
1207 stack_depth: 0,
1208 stale_from_parent: false,
1209 task: None,
1210 task_assignment_id: None,
1211 task_summary: None,
1212 changed_paths: vec![],
1213 promotion_suggested: false,
1214 impact_categories: vec![],
1215 heavy_impact_paths: vec![],
1216 verification_summary: Default::default(),
1217 confidence_summary: Default::default(),
1218 integration_policy_result: Default::default(),
1219 coordination_status: CoordinationStatus::Clean,
1220 is_current: false,
1221 is_isolated: false,
1222 thread_health: "clean".into(),
1223 blockers: vec![],
1224 recommended_action: String::new(),
1225 recommended_action_template: None,
1226 git_branch_tip: None,
1227 history_imported: true,
1228 auto: false,
1229 shared_target_dir: None,
1230 },
1231 ThreadListEntry {
1232 name: "origin/main".into(),
1233 operation: None,
1234 remote_tracking: None,
1235 base_state: None,
1236 base_root: None,
1237 current_state: None,
1238 path: None,
1239 execution_path: None,
1240 session_id: None,
1241 heddle_session_id: None,
1242 actor: None,
1243 harness: None,
1244 thinking_level: None,
1245 native_actor_key: None,
1246 native_parent_actor_key: None,
1247 probe_source: None,
1248 probe_confidence: None,
1249 usage_summary: None,
1250 last_progress_at: None,
1251 last_activity_at: None,
1252 report_flush_state: None,
1253 attach_reason: None,
1254 thread_mode: None,
1255 thread_state: None,
1256 freshness: None,
1257 visibility: "ref_only".into(),
1258 target_thread: None,
1259 parent_thread: None,
1260 child_threads: vec![],
1261 sibling_threads: vec![],
1262 stack_depth: 0,
1263 stale_from_parent: false,
1264 task: None,
1265 task_assignment_id: None,
1266 task_summary: None,
1267 changed_paths: vec![],
1268 promotion_suggested: false,
1269 impact_categories: vec![],
1270 heavy_impact_paths: vec![],
1271 verification_summary: Default::default(),
1272 confidence_summary: Default::default(),
1273 integration_policy_result: Default::default(),
1274 coordination_status: CoordinationStatus::Clean,
1275 is_current: false,
1276 is_isolated: false,
1277 thread_health: "git_backed".into(),
1278 blockers: vec![],
1279 recommended_action: "heddle thread switch origin/main".into(),
1280 recommended_action_template: None,
1281 git_branch_tip: Some("deadbeef".into()),
1282 history_imported: false,
1283 auto: false,
1284 shared_target_dir: None,
1285 },
1286 ];
1287
1288 let available = split_available_git_refs(&mut summaries);
1289 assert_eq!(summaries.len(), 1);
1290 assert_eq!(summaries[0].name, "feature");
1291 assert_eq!(available.len(), 1);
1292 assert_eq!(available[0].name, "origin/main");
1293 assert_eq!(available[0].git_commit, "deadbeef");
1294 }
1295}