use std::time::SystemTime;
use anyhow::{Context, Result, anyhow};
pub(crate) use heddle_cli_contract::cli::commands::wire::collab::{
AnchorOutput, DiscussWaitLineOutput, DiscussionListOutput, DiscussionOutput,
DiscussionShowOutput, DiscussionWriteOutput, ResolutionOutput, TurnOutput,
};
use objects::object::{
AnnotationKind, CollabOpId, CollaborationAnchor, CollaborationAnchorStatus,
CollaborationIdempotencyKey, CollaborationOperationBodyV1, CollaborationOperationEnvelope,
CollaborationResolution, DiscussionRecordId, DiscussionTurnV1, MaterializedDiscussion, StateId,
VisibilityTier,
};
use repo::{
CollaborationStore, CollaborationWriteDisposition, CollaborationWriteOutcome,
RepositoryCapability, migrate_legacy_discussions_once,
};
use super::{
advice::RecoveryAdvice,
compact::{CompactOutput, CompactProjection},
history_target::resolve_state_id,
native_scope::{
AnnotationStore, AnnotationSurface, emit_locality_notice_once, open_annotation_store,
report_absent_store,
},
next_action::{
NextActionValidationContext, write_full_command_json, write_projected_command_json,
},
snapshot::ensure_current_state,
};
use crate::{
cli::{
cli_args::{
Cli, DiscussAppendArgs, DiscussCommands, DiscussListArgs, DiscussOpenArgs,
DiscussReopenArgs, DiscussResolveArgs, DiscussShowArgs, DiscussWaitArgs,
ResolveModeArg,
},
should_output_json,
},
config::UserConfig,
};
pub async fn run(cli: &Cli, command: &DiscussCommands) -> Result<()> {
let repo = match command {
DiscussCommands::List(_) | DiscussCommands::Show(_) => match open_annotation_store(cli)? {
AnnotationStore::Present(repo) => *repo,
AnnotationStore::Absent(absent) => {
let output_kind = match command {
DiscussCommands::Show(_) => "discuss_show",
_ => "discuss_list",
};
let with_items = matches!(command, DiscussCommands::List(_));
return report_absent_store(
cli,
AnnotationSurface::Discuss,
output_kind,
with_items,
&absent,
);
}
},
#[cfg(feature = "client")]
DiscussCommands::Wait(args) => {
let repo = cli.open_repo().context("open Heddle repository")?;
return run_wait(cli, &repo, args).await;
}
#[cfg(not(feature = "client"))]
DiscussCommands::Wait(_) => {
return Err(anyhow!("discuss wait requires the hosted client feature"));
}
_ => cli.open_repo().context("open Heddle repository")?,
};
let store = if migrates_legacy_on_entry(command) {
open_store(&repo)?
} else {
CollaborationStore::open(repo.heddle_dir()).context("open collaboration store")?
};
match command {
DiscussCommands::Open(args) => run_open(cli, &repo, &store, args),
DiscussCommands::Append(args) => run_append(cli, &repo, &store, args),
DiscussCommands::Resolve(args) => run_resolve(cli, &repo, &store, args),
DiscussCommands::Reopen(args) => run_reopen(cli, &repo, &store, args),
DiscussCommands::List(args) => run_list(cli, &repo, &store, args),
DiscussCommands::Show(args) => run_show(cli, &store, args),
DiscussCommands::Wait(_) => unreachable!("wait returns before opening the store"),
}
}
fn migrates_legacy_on_entry(command: &DiscussCommands) -> bool {
!matches!(command, DiscussCommands::Wait(_))
}
impl CompactProjection for DiscussionWriteOutput {
fn compact(&self) -> CompactOutput {
let mut compact = CompactOutput::new(self.output_kind);
compact.status = Some(match self.disposition {
CollaborationWriteDisposition::Created => "created".to_string(),
CollaborationWriteDisposition::ExistingOperation => "existing_operation".to_string(),
CollaborationWriteDisposition::IdempotentReplay => "idempotent_replay".to_string(),
});
compact
}
}
fn open_store(repo: &repo::Repository) -> Result<CollaborationStore> {
let store = CollaborationStore::open(repo.heddle_dir()).context("open collaboration store")?;
migrate_legacy_discussions_once(repo, &store, repo.get_attribution()?)
.context("migrate legacy discussions")?;
Ok(store)
}
fn run_open(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
args: &DiscussOpenArgs,
) -> Result<()> {
let (file, symbol, body) = open_inputs(args)?;
let state_id = resolve_open_state(repo, args.state.as_deref())?;
let title = args
.title
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.or_else(|| body.lines().map(str::trim).find(|line| !line.is_empty()))
.unwrap_or(symbol)
.to_string();
let discussion_id = DiscussionRecordId::generate();
let operation = CollaborationOperationEnvelope::new(
discussion_id,
Vec::new(),
idempotency_key(cli)?,
repo.get_attribution()?,
now_ms(),
CollaborationOperationBodyV1::Open {
title,
anchor: CollaborationAnchor::Symbol {
state_id,
path: file.to_string(),
symbol: symbol.to_string(),
},
visibility: parse_visibility(
args.visibility.as_deref(),
repo.resolve_capture_default_visibility(),
)?,
turn: DiscussionTurnV1::new(body)?,
thread_ref: args.thread.clone(),
},
)?;
let outcome = store.write_operation(&operation)?;
emit_locality_notice_once(repo, AnnotationSurface::Discuss);
emit_write(cli, "discuss_open", store, discussion_id, outcome)
}
fn open_inputs(args: &DiscussOpenArgs) -> Result<(&str, &str, &str)> {
match (
args.file.as_deref(),
args.symbol.as_deref(),
args.body.as_deref(),
args.file_flag.as_deref(),
args.symbol_flag.as_deref(),
args.body_flag.as_deref(),
) {
(Some(file), Some(symbol), Some(body), None, None, None)
| (None, None, None, Some(file), Some(symbol), Some(body)) => Ok((file, symbol, body)),
_ => Err(anyhow!(
"discuss open requires either <FILE> <SYMBOL> <BODY> or --file <FILE> --symbol <SYMBOL> --body <BODY>"
)),
}
}
fn run_append(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
args: &DiscussAppendArgs,
) -> Result<()> {
write_descendant(
cli,
repo,
store,
&args.discussion_id,
"discuss_append",
CollaborationOperationBodyV1::AppendTurn {
turn: DiscussionTurnV1::new(args.body.clone())?,
},
)
}
fn run_resolve(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
args: &DiscussResolveArgs,
) -> Result<()> {
let resolution = match (args.mode.as_ref(), args.into_annotation) {
(Some(ResolveModeArg::ByEdit), false) => CollaborationResolution::AddressedByState {
state_id: resolve_state(repo, args.state.as_deref())?,
},
(Some(ResolveModeArg::Dismiss), false) => CollaborationResolution::Dismissed {
reason: args
.reason
.as_deref()
.map(str::trim)
.filter(|reason| !reason.is_empty())
.ok_or_else(|| anyhow!(RecoveryAdvice::discuss_resolve_missing_dismiss_reason()))?
.to_string(),
},
(None, true) => CollaborationResolution::IntoAnnotation {
annotation_kind: args
.kind
.as_deref()
.unwrap_or("rationale")
.parse::<AnnotationKind>()
.map_err(|error| anyhow!(error))?,
content: args
.body
.as_deref()
.map(str::trim)
.filter(|body| !body.is_empty())
.ok_or_else(|| anyhow!("--body must not be empty for --into-annotation"))?
.to_string(),
tags: args.tag.clone(),
},
_ => {
return Err(anyhow!(
"discuss resolve requires exactly one of --mode or --into-annotation"
));
}
};
write_descendant(
cli,
repo,
store,
&args.discussion_id,
"discuss_resolve",
CollaborationOperationBodyV1::Resolve { resolution },
)
}
fn run_reopen(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
args: &DiscussReopenArgs,
) -> Result<()> {
write_descendant(
cli,
repo,
store,
&args.discussion_id,
"discuss_reopen",
CollaborationOperationBodyV1::Reopen {
reason: args.reason.clone(),
},
)
}
fn write_descendant(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
raw_id: &str,
output_kind: &'static str,
body: CollaborationOperationBodyV1,
) -> Result<()> {
let discussion_id = parse_discussion_id(raw_id)?;
let discussion = store
.materialize_discussion(&discussion_id)?
.ok_or_else(|| anyhow!("discussion {discussion_id} not found"))?;
let operation = CollaborationOperationEnvelope::new(
discussion_id,
discussion.heads.iter().copied().collect(),
idempotency_key(cli)?,
repo.get_attribution()?,
now_ms(),
body,
)?;
let outcome = store.write_operation(&operation)?;
emit_write(cli, output_kind, store, discussion_id, outcome)
}
fn run_list(
cli: &Cli,
repo: &repo::Repository,
store: &CollaborationStore,
args: &DiscussListArgs,
) -> Result<()> {
if args.symbol.is_some() && args.file.is_none() {
return Err(anyhow!("discuss list --symbol requires --file"));
}
if !matches!(
args.status.as_str(),
"all" | "open" | "resolved" | "conflicted"
) {
return Err(anyhow!(
"invalid discussion status {:?}; expected open, resolved, conflicted, or all",
args.status
));
}
let state_filter = args
.state
.as_deref()
.map(|value| resolve_state(repo, Some(value)))
.transpose()?;
let materialized = store.materialize()?;
let mut discussions = Vec::new();
for discussion in materialized.discussions.into_values() {
if !matches_filters(&discussion, args, state_filter.as_ref()) {
continue;
}
discussions.push(to_view(store, &discussion)?);
}
let output = DiscussionListOutput {
output_kind: "discuss_list",
discussions,
};
if should_output_json(cli, None) {
write_full_command_json(
&output,
NextActionValidationContext::without_repo(&["discuss", "list"]),
)?;
} else if output.discussions.is_empty() {
println!("(no discussions)");
} else {
for discussion in output.discussions {
println!(
"{} [{}] {} — {}",
discussion.id,
discussion.status,
anchor_label(&discussion.anchor),
discussion.title
);
}
}
Ok(())
}
#[cfg(feature = "client")]
async fn run_wait(cli: &Cli, repo: &repo::Repository, args: &DiscussWaitArgs) -> Result<()> {
use hosted_client::client::discussion_live::{
DiscussionCursorScope, DiscussionEventConsumer, DiscussionEventOutcome, load_scoped_cursor,
paired_thread_scope, save_scoped_cursor, wait_reconnect_backoff,
};
use hosted_client::client::{HostedAuthMode, HostedClient};
use super::remote::resolve_default_remote_name;
use crate::remote::{RemoteTarget, resolve_remote_with_key_and_insecure};
let remote_name = resolve_default_remote_name(repo, args.remote.as_deref())?;
let (target, server_key, insecure) =
resolve_remote_with_key_and_insecure(repo, Some(&remote_name))?;
let (authority, repo_path) = match target {
RemoteTarget::Network {
authority,
repo_path,
} => (
authority,
repo_path.context("hosted remote must include a repository path")?,
),
RemoteTarget::Local(_) => {
return Err(anyhow!(RecoveryAdvice::safety_refusal(
"hosted_remote_required",
format!("discuss wait requires a hosted remote; remote '{remote_name}' is local"),
"Configure a hosted remote, then retry `heddle discuss wait`.",
format!("remote '{remote_name}' is local, but live discussion delivery is hosted"),
"a local remote has no event cursor",
"no hosted request was sent and local repository state was left unchanged",
"heddle remote list",
vec!["heddle remote list".to_string()],
)));
}
};
let user_config = UserConfig::load_default()?;
let mut client = HostedClient::open_session_with_insecure(
&authority,
&user_config,
server_key.clone(),
HostedAuthMode::CredentialFallback,
insecure,
)
.await?;
let (thread_name, thread_id) = if let Some(thread) = args.thread.as_deref() {
let record = super::thread_cmd::load_thread(repo, thread)?;
paired_thread_scope(&record.thread, &record.id)?
} else {
(String::new(), String::new())
};
let cursor_scope = DiscussionCursorScope {
authority: authority.clone(),
repo_path: repo_path.clone(),
thread: thread_name.clone(),
thread_id: thread_id.clone(),
principal: client.authenticated_username().unwrap_or_default(),
};
if let Some(after) = args.after {
let mut cursor = load_scoped_cursor(repo.heddle_dir(), &cursor_scope)?;
cursor.after_event_id = after;
save_scoped_cursor(repo.heddle_dir(), &cursor_scope, &cursor)?;
}
let mut seen = 0usize;
let mut reconnects = 0u32;
'session: loop {
let mut consumer = DiscussionEventConsumer::new(repo, &mut client, repo_path.clone())
.with_authority(authority.clone());
if !thread_name.is_empty() {
consumer = consumer.with_thread(&thread_name, &thread_id);
}
let mut subscription = consumer.start(None).await?;
emit_wait_line(
cli,
DiscussWaitLineOutput {
output_kind: "discuss_wait",
status: "listening",
event_id: subscription.last_event_id(),
event_type: String::new(),
discussion_id: None,
applied: false,
after_event_id: subscription.last_event_id(),
skip_reason: None,
},
)?;
loop {
if args.max_events.is_some_and(|max| seen >= max) {
break 'session;
}
match consumer.consume_next(&mut subscription).await {
Ok((event, outcome)) => {
reconnects = 0;
seen += 1;
let status = match &outcome {
DiscussionEventOutcome::Applied { .. } => "applied",
DiscussionEventOutcome::Unchanged { .. } => "unchanged",
DiscussionEventOutcome::Skipped { .. } => "skipped",
DiscussionEventOutcome::Ignored => "ignored",
};
emit_wait_line(
cli,
DiscussWaitLineOutput {
output_kind: "discuss_wait",
status,
event_id: event.event_id,
event_type: event.event_type,
discussion_id: outcome.discussion_id().map(ToString::to_string),
applied: outcome.applied(),
after_event_id: subscription.last_event_id(),
skip_reason: outcome.skip_reason().map(ToString::to_string),
},
)?;
if args.max_events.is_some_and(|max| seen >= max) {
break 'session;
}
}
Err(error) if error.resume_after_event_id().is_some() => {
let Some(delay) = wait_reconnect_backoff(reconnects) else {
return Err(anyhow!(
"discuss wait reconnect ceiling reached after {reconnects} attempts"
));
};
reconnects += 1;
drop(consumer);
tokio::time::sleep(delay).await;
client.close().await;
client = HostedClient::open_session_with_insecure(
&authority,
&user_config,
server_key.clone(),
HostedAuthMode::CredentialFallback,
insecure,
)
.await?;
continue 'session;
}
Err(error) => return Err(error.into()),
}
}
}
client.close().await;
Ok(())
}
#[cfg(feature = "client")]
fn emit_wait_line(cli: &Cli, line: DiscussWaitLineOutput) -> Result<()> {
if should_output_json(cli, None) {
println!("{}", serde_json::to_string(&line)?);
return Ok(());
}
match line.status {
"listening" => println!(
"waiting for discussion events after {}",
line.after_event_id
),
"applied" => println!(
"applied {} {} ({})",
line.event_type,
line.discussion_id.as_deref().unwrap_or("-"),
line.event_id
),
"unchanged" => println!(
"already had {} {} ({})",
line.event_type,
line.discussion_id.as_deref().unwrap_or("-"),
line.event_id
),
"skipped" => println!("{}", format_wait_skip(&line)),
"ignored" => {}
other => println!("{other} {} ({})", line.event_type, line.event_id),
}
Ok(())
}
fn run_show(cli: &Cli, store: &CollaborationStore, args: &DiscussShowArgs) -> Result<()> {
let discussion_id = parse_discussion_id(&args.discussion_id)?;
let discussion = store
.materialize_discussion(&discussion_id)?
.ok_or_else(|| anyhow!("discussion {discussion_id} not found"))?;
let output = DiscussionShowOutput {
output_kind: "discuss_show",
discussion: to_view(store, &discussion)?,
};
emit_show(cli, &output)
}
fn emit_write(
cli: &Cli,
output_kind: &'static str,
store: &CollaborationStore,
discussion_id: DiscussionRecordId,
outcome: CollaborationWriteOutcome,
) -> Result<()> {
let discussion = store
.materialize_discussion(&discussion_id)?
.ok_or_else(|| anyhow!("discussion {discussion_id} was not materialized after write"))?;
let output = DiscussionWriteOutput {
output_kind,
operation_id: outcome.operation_id.to_string_full(),
disposition: outcome.disposition,
discussion: to_view(store, &discussion)?,
};
if should_output_json(cli, None) {
let emitting = match output_kind {
"discuss_open" => &["discuss", "open"][..],
"discuss_append" => &["discuss", "append"][..],
"discuss_resolve" => &["discuss", "resolve"][..],
"discuss_reopen" => &["discuss", "reopen"][..],
_ => &["discuss"][..],
};
write_projected_command_json(cli, &output, emitting)?;
} else {
println!(
"{} {} ({})",
output.discussion.status, output.discussion.id, output.operation_id
);
}
Ok(())
}
fn emit_show(cli: &Cli, output: &DiscussionShowOutput) -> Result<()> {
if should_output_json(cli, None) {
write_full_command_json(
output,
NextActionValidationContext::without_repo(&["discuss", "show"]),
)?;
return Ok(());
}
let discussion = &output.discussion;
println!("discussion {} [{}]", discussion.id, discussion.status);
println!(" title: {}", discussion.title);
let anchor_status = match discussion.anchor_status {
"current" => String::new(),
status => format!(" [{status}]"),
};
println!(
" anchor: {}{}",
anchor_label(&discussion.anchor),
anchor_status
);
println!(" visibility: {}", discussion.visibility);
if let Some(thread_ref) = &discussion.thread_ref {
println!(" thread: {thread_ref}");
}
println!(" heads: {}", discussion.head_operation_ids.join(", "));
for turn in &discussion.turns {
let actor = turn.agent.as_deref().unwrap_or(&turn.author_name);
println!(
" {} {} @ {}",
turn.operation_id, actor, turn.occurred_at_ms
);
for line in turn.body.lines() {
println!(" {line}");
}
}
Ok(())
}
fn to_view(store: &CollaborationStore, value: &MaterializedDiscussion) -> Result<DiscussionOutput> {
let mut turns = Vec::with_capacity(value.turns.len());
for (operation_id, turn) in &value.turns {
let decoded = store
.read_operation(operation_id)?
.ok_or_else(|| anyhow!("discussion references missing operation {operation_id}"))?;
let author = decoded.operation.author;
turns.push(TurnOutput {
operation_id: operation_id.to_string_full(),
author_name: author.principal.name_lossy().into_owned(),
author_email: author.principal.email_lossy().into_owned(),
agent: author.agent.map(|agent| agent.to_string()),
occurred_at_ms: decoded.operation.occurred_at_ms,
body: turn.body.clone(),
content_hash: turn.content_hash.to_hex(),
});
}
let status = if !value.conflict_operations.is_empty() {
"conflicted"
} else if value.resolution.is_some() {
"resolved"
} else {
"open"
};
Ok(DiscussionOutput {
id: value.discussion_id.to_string(),
title: value.title.clone(),
anchor: anchor_output(&value.anchor),
anchor_status: anchor_status_token(value.anchor_status),
visibility: visibility_token(&value.visibility),
thread_ref: value.thread_ref.clone(),
status,
resolution: value.resolution.as_ref().map(resolution_output),
conflict_operation_ids: value
.conflict_operations
.iter()
.map(CollabOpId::to_string_full)
.collect(),
head_operation_ids: value.heads.iter().map(CollabOpId::to_string_full).collect(),
display_head_operation_id: value.display_head.to_string_full(),
turns,
})
}
fn anchor_status_token(status: CollaborationAnchorStatus) -> &'static str {
match status {
CollaborationAnchorStatus::Current => "current",
CollaborationAnchorStatus::Moved => "moved",
CollaborationAnchorStatus::Ambiguous => "ambiguous",
CollaborationAnchorStatus::Orphaned => "orphaned",
}
}
fn matches_filters(
discussion: &MaterializedDiscussion,
args: &DiscussListArgs,
state: Option<&StateId>,
) -> bool {
let status_matches = match args.status.as_str() {
"open" => discussion.resolution.is_none() && discussion.conflict_operations.is_empty(),
"resolved" => discussion.resolution.is_some() && discussion.conflict_operations.is_empty(),
"conflicted" => !discussion.conflict_operations.is_empty(),
_ => true,
};
status_matches
&& state.is_none_or(|state| anchor_state(&discussion.anchor) == Some(state))
&& args.file.as_ref().is_none_or(|path| {
anchor_path(&discussion.anchor).is_some_and(|candidate| candidate == path)
})
&& args.symbol.as_ref().is_none_or(|symbol| {
matches!(&discussion.anchor, CollaborationAnchor::Symbol { symbol: candidate, .. } if candidate == symbol)
})
}
fn parse_discussion_id(value: &str) -> Result<DiscussionRecordId> {
value
.parse()
.map_err(|error| anyhow!("invalid discussion id {value:?}: {error}"))
}
fn idempotency_key(cli: &Cli) -> Result<CollaborationIdempotencyKey> {
CollaborationIdempotencyKey::new(
cli.op_id
.clone()
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
)
.map_err(anyhow::Error::msg)
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or(0)
}
fn resolve_state(repo: &repo::Repository, explicit: Option<&str>) -> Result<StateId> {
if let Some(value) = explicit {
return resolve_state_id(repo, value);
}
repo.head()?
.ok_or_else(|| anyhow!(RecoveryAdvice::repository_no_head_anchor_first("discuss")))
}
fn resolve_open_state(repo: &repo::Repository, explicit: Option<&str>) -> Result<StateId> {
if explicit.is_some() || repo.head()?.is_some() {
return resolve_state(repo, explicit);
}
if repo.capability() == RepositoryCapability::GitOverlay
&& repo
.git_overlay_worktree_status()?
.is_some_and(|status| status.is_clean())
{
return ensure_current_state(
repo,
&UserConfig::load_default()?,
Some("Bootstrap git-overlay before opening discussion".to_string()),
);
}
Err(anyhow!(RecoveryAdvice::repository_no_head_anchor_first(
"discuss"
)))
}
fn parse_visibility(value: Option<&str>, default: VisibilityTier) -> Result<VisibilityTier> {
let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else {
return Ok(default);
};
match value {
"public" => Ok(VisibilityTier::Public),
"internal" => Ok(VisibilityTier::Internal),
_ if value.starts_with("team:") => labelled_visibility(value, "team:", |label| {
VisibilityTier::TeamScoped { team_id: label }
}),
_ if value.starts_with("restricted:") => {
labelled_visibility(value, "restricted:", |label| VisibilityTier::Restricted {
scope_label: label,
})
}
_ if value.starts_with("private:") => labelled_visibility(value, "private:", |label| {
VisibilityTier::Private { scope_label: label }
}),
_ => Err(anyhow!(
"invalid visibility {value:?}; expected public, internal, team:<id>, restricted:<label>, or private:<label>"
)),
}
}
fn labelled_visibility(
value: &str,
prefix: &str,
build: impl FnOnce(String) -> VisibilityTier,
) -> Result<VisibilityTier> {
let label = value.trim_start_matches(prefix).trim();
if label.is_empty() {
return Err(anyhow!("visibility {prefix}<label> requires a label"));
}
Ok(build(label.to_string()))
}
fn visibility_token(value: &VisibilityTier) -> String {
match value {
VisibilityTier::Public => "public".to_string(),
VisibilityTier::Internal => "internal".to_string(),
VisibilityTier::TeamScoped { team_id } => format!("team:{team_id}"),
VisibilityTier::Restricted { scope_label } => format!("restricted:{scope_label}"),
VisibilityTier::Private { scope_label } => format!("private:{scope_label}"),
}
}
fn anchor_output(value: &CollaborationAnchor) -> AnchorOutput {
match value {
CollaborationAnchor::Repository => AnchorOutput::Repository,
CollaborationAnchor::State { state_id } => AnchorOutput::State {
state_id: state_id.to_string_full(),
},
CollaborationAnchor::Change { change_id } => AnchorOutput::Change {
change_id: change_id.to_string_full(),
},
CollaborationAnchor::Path { state_id, path } => AnchorOutput::Path {
state_id: state_id.to_string_full(),
path: path.clone(),
},
CollaborationAnchor::Symbol {
state_id,
path,
symbol,
} => AnchorOutput::Symbol {
state_id: state_id.to_string_full(),
path: path.clone(),
symbol: symbol.clone(),
},
}
}
fn resolution_output(value: &CollaborationResolution) -> ResolutionOutput {
match value {
CollaborationResolution::AddressedByState { state_id } => {
ResolutionOutput::AddressedByState {
state_id: state_id.to_string_full(),
}
}
CollaborationResolution::AddressedByChange { change_id } => {
ResolutionOutput::AddressedByChange {
change_id: change_id.to_string_full(),
}
}
CollaborationResolution::Dismissed { reason } => ResolutionOutput::Dismissed {
reason: reason.clone(),
},
CollaborationResolution::IntoAnnotation {
annotation_kind,
content,
tags,
} => ResolutionOutput::IntoAnnotation {
annotation_kind: annotation_kind.to_string(),
content: content.clone(),
tags: tags.clone(),
},
CollaborationResolution::Annotation { annotation_id } => ResolutionOutput::Annotation {
annotation_id: annotation_id.clone(),
},
}
}
fn anchor_state(value: &CollaborationAnchor) -> Option<&StateId> {
match value {
CollaborationAnchor::State { state_id }
| CollaborationAnchor::Path { state_id, .. }
| CollaborationAnchor::Symbol { state_id, .. } => Some(state_id),
CollaborationAnchor::Repository | CollaborationAnchor::Change { .. } => None,
}
}
fn anchor_path(value: &CollaborationAnchor) -> Option<&str> {
match value {
CollaborationAnchor::Path { path, .. } | CollaborationAnchor::Symbol { path, .. } => {
Some(path)
}
_ => None,
}
}
fn format_wait_skip(line: &DiscussWaitLineOutput) -> String {
match line
.skip_reason
.as_deref()
.filter(|reason| !reason.is_empty())
{
Some(reason) => format!("skipped {} ({}) — {reason}", line.event_type, line.event_id),
None => format!("skipped {} ({})", line.event_type, line.event_id),
}
}
fn anchor_label(value: &AnchorOutput) -> String {
match value {
AnchorOutput::Repository => "repository".to_string(),
AnchorOutput::State { state_id } => state_id.clone(),
AnchorOutput::Change { change_id } => change_id.clone(),
AnchorOutput::Path { path, .. } => path.clone(),
AnchorOutput::Symbol { path, symbol, .. } => format!("{path}:{symbol}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cli::cli_args::{DiscussListArgs, DiscussShowArgs, DiscussWaitArgs};
#[test]
fn wait_is_the_only_discuss_verb_that_skips_legacy_migration() {
let wait = DiscussCommands::Wait(DiscussWaitArgs {
after: None,
remote: None,
thread: None,
max_events: None,
});
assert!(
!migrates_legacy_on_entry(&wait),
"discuss wait must not migrate legacy discussions before hosted bootstrap"
);
let list = DiscussCommands::List(DiscussListArgs {
state: None,
file: None,
symbol: None,
status: "open".to_string(),
});
assert!(
migrates_legacy_on_entry(&list),
"discuss list must still convert legacy attachments"
);
let show = DiscussCommands::Show(DiscussShowArgs {
discussion_id: "disc-1".to_string(),
});
assert!(
migrates_legacy_on_entry(&show),
"discuss show must still convert legacy attachments"
);
}
#[test]
fn list_and_show_materialize_legacy_attachments() {
use chrono::Utc;
use objects::{
object::{
Attribution, Blob, Discussion, DiscussionResolution, DiscussionTurn,
DiscussionsBlob, Principal, StateAttachment, StateAttachmentBody, SymbolAnchor,
},
store::ObjectStore,
};
let temp = tempfile::TempDir::new().unwrap();
let repo = repo::Repository::init_default(temp.path()).unwrap();
let state_id = repo.head().unwrap().unwrap();
let bytes = DiscussionsBlob::new(vec![Discussion {
id: "legacy-1".to_string(),
anchor: SymbolAnchor::new("src/lib.rs", "run"),
opened_against_state: state_id,
opened_at: 1_700_000_000,
thread_ref: None,
turns: vec![DiscussionTurn {
author: Principal::new("Ada", "ada@example.com"),
body: "why?".to_string(),
posted_at: 1_700_000_000,
references: Vec::new(),
}],
resolution: DiscussionResolution::Open,
body_changed_since_open: false,
anchor_ambiguous: false,
orphaned: false,
visibility: VisibilityTier::default(),
resolved_annotation_id: None,
}])
.encode()
.unwrap();
let blob_hash = repo.store().put_blob(&Blob::new(bytes)).unwrap();
repo.put_state_attachment(&StateAttachment {
state_id,
body: StateAttachmentBody::Discussions(blob_hash),
attribution: Attribution::human(Principal::new("Importer", "importer@example.com")),
created_at: Utc::now(),
supersedes: None,
})
.unwrap();
let marker = repo
.heddle_dir()
.join("collaboration/migrations/legacy-discussions-v1");
assert!(
!marker.exists(),
"the fixture is an unmigrated clone with legacy attachments"
);
let store = open_store(&repo).expect("list/show open the store through migrate");
let materialized = store.materialize().unwrap();
assert_eq!(
materialized.discussions.len(),
1,
"list/show must still convert legacy attachments when the marker is unset"
);
let discussion = materialized.discussions.into_values().next().unwrap();
assert_eq!(discussion.turns[0].1.body, "why?");
assert!(
marker.exists(),
"list/show claim the marker after converting the attachment"
);
assert!(
store
.materialize_discussion(&discussion.discussion_id)
.unwrap()
.is_some(),
"discuss show must be able to load the migrated discussion by id"
);
}
#[test]
fn skipped_wait_line_carries_the_reason() {
let line = DiscussWaitLineOutput {
output_kind: "discuss_wait",
status: "skipped",
event_id: 44,
event_type: "discussion.opened".to_string(),
discussion_id: None,
applied: false,
after_event_id: 44,
skip_reason: Some("discussion disc-hidden is not visible to this caller".to_string()),
};
let json = serde_json::to_value(&line).unwrap();
assert_eq!(
json["skip_reason"],
"discussion disc-hidden is not visible to this caller"
);
assert!(
format_wait_skip(&line).contains("not visible"),
"human skip lines must include the reason"
);
}
}