From 412dc53e38b02d5bad8fa128f002b7ce3ccc4b65 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 00:03:33 +0000 Subject: [PATCH] fix(buzz): make the workflow + job boards actually show runs (aggregate from the live subscription) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The boards derived their state from LocalCache.filter(kinds = 46xxx / 43xxx), but LocalCache.filter only matches notes whose kind.isRegular() (< 10_000). Every run/lifecycle/job kind is >= 43001, so the filter returned nothing and the boards never displayed a single run/job against live data (verified with a probe: a consumed 46020 matched 0, a 30620 def matched 1). Definitions (30620, addressable) were the only thing that showed. Aggregate straight off subscribeAsFlow, which accumulates the channel's stored + live events (deduped by id) and re-emits the list — the data the aggregators need. This also removes the per-batch whole-cache rescans. - WorkflowRunBoardViewModel: base #h subscription + a nested by-author decisions subscription (rebuilt only when the approver set changes via distinctUntilChanged) so grant/deny now arrive live for every observer, not just once at open. Drop 46004/46011/46012 from the fetch set — the aggregator can't correlate them. - JobBoardViewModel: aggregate jobs + kind-7 upvotes from the one #h subscription. - Real success/failure feedback: trigger/approve/deny/defineWorkflow return a result; snackbar only on confirmed publish; the sheet stays open on a failed trigger; the definition editor shows an error + a Publishing… state instead of hanging open and inviting duplicate 30620s. - Gate write actions on isWriteable(): a read-only login no longer sees a false "Approved" success, and the New-run FAB is hidden. - WorkflowRunAggregator.fold: parse each event's JSON content once. - Empty-state hint in the New-run sheet when no definitions exist yet. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011mApqAbr8vkLC7gUDjavu6 --- .../screen/loggedIn/buzz/JobBoardViewModel.kt | 80 +++--- .../loggedIn/buzz/WorkflowRunBoardScreen.kt | 127 +++++++--- .../buzz/WorkflowRunBoardViewModel.kt | 239 +++++++++--------- .../commons/model/buzz/WorkflowRuns.kt | 6 +- 4 files changed, 251 insertions(+), 201 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/JobBoardViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/JobBoardViewModel.kt index 5c45c3f4e9..d634500dd6 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/JobBoardViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/JobBoardViewModel.kt @@ -25,8 +25,6 @@ import androidx.lifecycle.viewModelScope import com.vitorpamplona.amethyst.commons.model.buzz.BuzzJobAggregator import com.vitorpamplona.amethyst.commons.model.buzz.JobView import com.vitorpamplona.amethyst.model.Account -import com.vitorpamplona.amethyst.model.LocalCache -import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllWithHooks import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.subscribeAsFlow import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl @@ -34,24 +32,27 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.onStart import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock /** * Backing ViewModel for the [JobBoardScreen] — the shared backlog of one Buzz channel. * * A Buzz "job" is the agent-job protocol (kinds 43001-43006): a member files a request, the * workspace bot works it, and every step lands as a signed event the whole room sees. This VM - * fetches those events (plus their kind-7 upvotes) scoped to the channel `h`, folds them into - * per-job [JobView] records via the shared [BuzzJobAggregator], and exposes the backlog as a - * [StateFlow]. It also drives the three write actions the board offers: file, upvote, cancel. + * folds those events (plus their kind-7 upvotes) into per-job [JobView] records via the shared + * [BuzzJobAggregator] and exposes the backlog as a [StateFlow]. It also drives the three write + * actions the board offers: file, upvote, cancel. * - * The heavy lifting (correlation, state machine, upvote priority) lives in `commons`; this VM is - * the Android glue (fetch → LocalCache → re-derive → publish via [Account]). + * **Aggregated from the live subscription, not `LocalCache`:** the job kinds 43001-43006 are neither + * regular (`< 10_000`) nor addressable, so `LocalCache.filter` can't serve them (its note branch only + * matches `kind.isRegular()`) — reading them back from cache returned nothing and the board stayed + * empty. We consume the events straight off [subscribeAsFlow], which accumulates the channel's stored + * + live events (deduped) and re-emits the list. The kind-7 upvotes ride the same `#h` subscription. */ class JobBoardViewModel : ViewModel() { @Volatile private var account: Account? = null @@ -65,7 +66,6 @@ class JobBoardViewModel : ViewModel() { val isLoading: StateFlow = _isLoading.asStateFlow() private var watchJob: Job? = null - private val reloadMutex = Mutex() fun bind( account: Account, @@ -76,29 +76,9 @@ class JobBoardViewModel : ViewModel() { this.account = account this.channelId = channelId this.relay = RelayUrlNormalizer.normalizeOrNull(relayUrl) - refresh() } - fun refresh() { - val account = account ?: return - val relay = relay ?: return - val channelId = channelId ?: return - viewModelScope.launch(Dispatchers.IO) { - _isLoading.value = true - try { - account.client.fetchAllWithHooks( - filters = mapOf(relay to boardFilters(channelId)), - timeoutMs = 8_000, - pendingOnAuthRequired = true, - ) { _, _ -> false } - reloadFromCache(channelId) - } finally { - _isLoading.value = false - } - } - } - - /** Keep the board live while it's on screen: any batch of job/reaction events re-derives it. */ + /** Keep the board live while it's on screen: the subscription backfills then streams job/reaction events. */ fun startWatching() { val account = account ?: return val relay = relay ?: return @@ -106,10 +86,22 @@ class JobBoardViewModel : ViewModel() { if (watchJob != null) return watchJob = viewModelScope.launch(Dispatchers.IO) { - account.client.subscribeAsFlow(relay, boardFilters(channelId)).collect { - // The client's global listener already consumed the batch into LocalCache. - reloadFromCache(channelId) - } + _isLoading.value = true + val loadingTimeout = + launch { + delay(LOADING_TIMEOUT_MS) + _isLoading.value = false + } + account.client + .subscribeAsFlow(relay, boardFilters(channelId)) + .onStart { emit(emptyList()) } + .collect { events -> + if (events.isNotEmpty()) { + loadingTimeout.cancel() + _isLoading.value = false + } + _jobs.value = BuzzJobAggregator.aggregate(events) + } } } @@ -118,15 +110,6 @@ class JobBoardViewModel : ViewModel() { watchJob = null } - private suspend fun reloadFromCache(channelId: String) = - reloadMutex.withLock { - val events = - LocalCache - .filter(Filter(kinds = ALL_KINDS, tags = mapOf("h" to listOf(channelId)))) - .mapNotNull { it.event } - _jobs.value = BuzzJobAggregator.aggregate(events) - } - fun file(request: String) = act { account, relay, channelId -> account.fileBuzzJob(relay, channelId, request) @@ -148,10 +131,8 @@ class JobBoardViewModel : ViewModel() { val account = account ?: return val relay = relay ?: return val channelId = channelId ?: return - viewModelScope.launch(Dispatchers.IO) { - block(account, relay, channelId) - reloadFromCache(channelId) // optimistic local re-derive; the live watch catches relay echoes - } + // The live subscription picks up the relay's echo of our own write, so no local re-derive here. + viewModelScope.launch(Dispatchers.IO) { block(account, relay, channelId) } } override fun onCleared() { @@ -160,10 +141,11 @@ class JobBoardViewModel : ViewModel() { } companion object { + private const val LOADING_TIMEOUT_MS = 6_000L + // The Buzz agent-job protocol (request/accepted/progress/result/cancel/error) plus the // kind-7 upvotes that prioritize it. private val JOB_KINDS = (43001..43006).toList() - private val ALL_KINDS = JOB_KINDS + ReactionEvent.KIND private fun boardFilters(channelId: String) = listOf( diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardScreen.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardScreen.kt index 30cdfa7015..428735675f 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardScreen.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardScreen.kt @@ -129,7 +129,8 @@ fun WorkflowRunBoardScreen( accountViewModel: AccountViewModel, nav: INav, ) { - val me = accountViewModel.account.userProfile().pubkeyHex + val me = remember(accountViewModel) { accountViewModel.account.userProfile().pubkeyHex } + val canWrite = remember(accountViewModel) { accountViewModel.account.isWriteable() } val viewModel: WorkflowRunBoardViewModel = viewModel(key = "WorkflowRunBoard-$relayUrl-$channelId") viewModel.bind(accountViewModel.account, channelId, relayUrl) @@ -152,11 +153,13 @@ fun WorkflowRunBoardScreen( topBar = { TopBarWithBackButton("Workflow runs", nav) }, snackbarHost = { SnackbarHost(snackbar) }, floatingActionButton = { - ExtendedFloatingActionButton( - onClick = { composing = true }, - icon = { Icon(symbol = MaterialSymbols.Add, contentDescription = null) }, - text = { Text("New run") }, - ) + if (canWrite) { + ExtendedFloatingActionButton( + onClick = { composing = true }, + icon = { Icon(symbol = MaterialSymbols.Add, contentDescription = null) }, + text = { Text("New run") }, + ) + } }, ) { padding -> Column(modifier = Modifier.padding(padding).fillMaxSize()) { @@ -173,12 +176,12 @@ fun WorkflowRunBoardScreen( verticalArrangement = Arrangement.spacedBy(12.dp), contentPadding = PaddingValues(top = 10.dp, bottom = 96.dp), ) { - section("Needs your approval", groups.awaiting, RunStyle.GATE, me, accountViewModel, nav) { run, grant -> + section("Needs your approval", groups.awaiting, RunStyle.GATE, me, canWrite, accountViewModel, nav) { run, grant -> pending = PendingDecision(run, grant) } - section("Working now", groups.active, RunStyle.ACTIVE, me, accountViewModel, nav) - section("Shipped", groups.done, RunStyle.SHIPPED, me, accountViewModel, nav) - section("Closed", groups.closed, RunStyle.CLOSED, me, accountViewModel, nav) + section("Working now", groups.active, RunStyle.ACTIVE, me, canWrite, accountViewModel, nav) + section("Shipped", groups.done, RunStyle.SHIPPED, me, canWrite, accountViewModel, nav) + section("Closed", groups.closed, RunStyle.CLOSED, me, canWrite, accountViewModel, nav) } } @@ -197,10 +200,18 @@ fun WorkflowRunBoardScreen( sheetState = sheetState, definitions = definitions, onDismiss = { composing = false }, - onDefine = { name, yaml, onCreated -> viewModel.defineWorkflow(name, yaml, onCreated) }, + onDefine = { name, yaml, onResult -> viewModel.defineWorkflow(name, yaml, onResult) }, onTrigger = { workflowId, task -> - viewModel.trigger(workflowId, task) - composing = false + // Only close the sheet on a confirmed publish; on failure keep it open (task text intact) + // and tell the user, instead of silently swallowing a read-only / rejected write. + viewModel.trigger(workflowId, task) { ok -> + if (ok) { + composing = false + scope.launch { snackbar.showSnackbar("Run triggered") } + } else { + scope.launch { snackbar.showSnackbar("Couldn't trigger the run — check you can post to this workspace") } + } + } }, ) } @@ -211,17 +222,20 @@ fun WorkflowRunBoardScreen( decision = decision, onDismiss = { pending = null }, onConfirm = { - if (decision.grant) viewModel.approve(decision.run.runId) else viewModel.deny(decision.run.runId) - pending = null - scope.launch { - snackbar.showSnackbar( - if (decision.grant) { - "Approved — the runner is opening a pull request" - } else { - "Denied — the work was discarded" - }, - ) + val grant = decision.grant + val onResult: (Boolean) -> Unit = { ok -> + scope.launch { + snackbar.showSnackbar( + when { + ok && grant -> "Approved — the runner is opening a pull request" + ok -> "Denied — the work was discarded" + else -> "Couldn't publish your decision — check you can post to this workspace" + }, + ) + } } + if (grant) viewModel.approve(decision.run.runId, onResult) else viewModel.deny(decision.run.runId, onResult) + pending = null }, ) } @@ -268,17 +282,18 @@ private fun LazyListScope.section( runs: List, style: RunStyle, me: String, + canWrite: Boolean, accountViewModel: AccountViewModel, nav: INav, onDecide: (WorkflowRun, Boolean) -> Unit = { _, _ -> }, ) { if (runs.isEmpty()) return - item(key = "header-$title") { + item(key = "header-$title", contentType = "header") { SectionHeader(title, runs.size, style) } - items(runs, key = { it.runId }) { run -> + items(runs, key = { it.runId }, contentType = { style }) { run -> if (style == RunStyle.GATE) { - GateCard(run, me, accountViewModel, nav, onDecide) + GateCard(run, me, canWrite, accountViewModel, nav, onDecide) } else { RunCard(run, style, accountViewModel, nav) } @@ -324,6 +339,7 @@ private fun SectionHeader( private fun LazyItemScope.GateCard( run: WorkflowRun, me: String, + canWrite: Boolean, accountViewModel: AccountViewModel, nav: INav, onDecide: (WorkflowRun, Boolean) -> Unit, @@ -383,11 +399,18 @@ private fun LazyItemScope.GateCard( Person("by", run.requester, accountViewModel, nav) } - if (mine) { + if (mine && canWrite) { ApprovalActions( onApprove = { onDecide(run, true) }, onDeny = { onDecide(run, false) }, ) + } else if (mine) { + // Named approver, but this login can't sign (read-only / remote signer w/o write). + Text( + text = "You're the approver, but this login can't sign a decision.", + style = MaterialTheme.typography.bodyMedium, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) } else { WaitingPill(run.pendingApprover, accountViewModel, nav) } @@ -675,7 +698,7 @@ private fun NewRunSheet( sheetState: SheetState, definitions: List, onDismiss: () -> Unit, - onDefine: (String, String, (String) -> Unit) -> Unit, + onDefine: (String, String, (String?) -> Unit) -> Unit, onTrigger: (String, String) -> Unit, ) { var task by remember { mutableStateOf("") } @@ -710,7 +733,12 @@ private fun NewRunSheet( if (defining) { DefinitionEditor( onCancel = { defining = false }, - onCreate = { name, yaml -> onDefine(name, yaml) { newId -> pendingSelectId = newId } }, + onCreate = { name, yaml, onResult -> + onDefine(name, yaml) { newId -> + pendingSelectId = newId // non-null → the effect below selects it and closes the editor + onResult(newId != null) // null → the editor shows its error and re-enables + } + }, ) } else { WorkflowPicker( @@ -719,6 +747,15 @@ private fun NewRunSheet( onSelect = { selected = it }, onNewDefinition = { defining = true }, ) + if (definitions.isEmpty()) { + // Don't leave a first-time user staring at a disabled Trigger button and an empty + // dropdown — point them at the way forward. + Text( + "No workflows yet. Open the menu above and choose “New definition…” to create one, then trigger it.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } OutlinedTextField( value = task, onValueChange = { task = it }, @@ -783,18 +820,24 @@ private fun WorkflowPicker( } } -/** Inline editor to publish a new kind-30620 definition: a name and its YAML recipe. */ +/** + * Inline editor to publish a new kind-30620 definition: a name and its YAML recipe. [onCreate] hands + * back a success flag; on failure the editor stays open, shows an error, and re-enables the button so + * the work isn't lost and the user isn't nudged into publishing a duplicate. + */ @Composable private fun DefinitionEditor( onCancel: () -> Unit, - onCreate: (String, String) -> Unit, + onCreate: (String, String, (Boolean) -> Unit) -> Unit, ) { var name by remember { mutableStateOf("") } var yaml by remember { mutableStateOf("") } + var publishing by remember { mutableStateOf(false) } + var error by remember { mutableStateOf(null) } Column(verticalArrangement = Arrangement.spacedBy(12.dp)) { Text("New workflow definition", style = MaterialTheme.typography.titleSmall, fontWeight = FontWeight.Bold) Text( - "Names it for the channel and publishes its YAML recipe (kind-30620). A real Buzz relay runs the YAML; self-hosted, the runner runs its configured command — the definition names and catalogs the run.", + "Names it for the channel and publishes its YAML recipe (kind-30620). A real Buzz relay runs the YAML. Self-hosted, the runner runs its configured command — here the definition just names and catalogs the run.", style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurfaceVariant, ) @@ -804,25 +847,37 @@ private fun DefinitionEditor( label = { Text("Name") }, placeholder = { Text("build-and-test") }, singleLine = true, + enabled = !publishing, modifier = Modifier.fillMaxWidth(), ) OutlinedTextField( value = yaml, onValueChange = { yaml = it }, label = { Text("YAML recipe") }, + enabled = !publishing, modifier = Modifier.fillMaxWidth(), minLines = 4, ) + error?.let { + Text(it, style = MaterialTheme.typography.bodyMedium, color = MaterialTheme.colorScheme.error) + } Row(modifier = Modifier.fillMaxWidth(), horizontalArrangement = Arrangement.spacedBy(10.dp)) { - OutlinedButton(onClick = onCancel) { Text("Cancel") } + OutlinedButton(onClick = onCancel, enabled = !publishing) { Text("Cancel") } Button( - onClick = { onCreate(name.trim(), yaml) }, - enabled = name.isNotBlank() && yaml.isNotBlank(), + onClick = { + error = null + publishing = true + onCreate(name.trim(), yaml) { ok -> + publishing = false + if (!ok) error = "Couldn't publish the definition — check you can post to this workspace." + } + }, + enabled = !publishing && name.isNotBlank() && yaml.isNotBlank(), modifier = Modifier.weight(1f), ) { Icon(symbol = MaterialSymbols.Add, contentDescription = null, modifier = Modifier.size(18.dp)) Spacer(Modifier.width(8.dp)) - Text("Create definition") + Text(if (publishing) "Publishing…" else "Create definition") } } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardViewModel.kt index be4eeb2de4..31b3145ee9 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardViewModel.kt @@ -26,35 +26,38 @@ import androidx.lifecycle.viewModelScope import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRun import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRunAggregator import com.vitorpamplona.amethyst.model.Account -import com.vitorpamplona.amethyst.model.LocalCache import com.vitorpamplona.quartz.buzz.workflow.ApprovalDenyEvent import com.vitorpamplona.quartz.buzz.workflow.ApprovalGrantEvent -import com.vitorpamplona.quartz.buzz.workflow.WorkflowApprovalDeniedEvent -import com.vitorpamplona.quartz.buzz.workflow.WorkflowApprovalGrantedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowApprovalRequestedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowCancelledEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowCompletedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowDefEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowFailedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowStepCompletedEvent -import com.vitorpamplona.quartz.buzz.workflow.WorkflowStepFailedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowStepStartedEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowTriggerEvent import com.vitorpamplona.quartz.buzz.workflow.WorkflowTriggeredEvent +import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey -import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllWithHooks import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.subscribeAsFlow import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.flatMapLatest +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.onStart import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext /** One selectable workflow definition (kind-30620) — its UUID `d` tag, optional name, and YAML recipe. */ @Immutable @@ -76,10 +79,19 @@ data class WorkflowDefOption( * This VM folds those events into per-run [WorkflowRun] records via the shared [WorkflowRunAggregator] * and exposes them, awaiting-approval first, as a [StateFlow]; it drives trigger + approve + deny. * - * Two fetch realities (both mirrored from `amy buzz workflow`): the trigger + lifecycle + gate are - * `#h`-scoped to the channel, but the grant/deny decisions carry only a `d` tag (= the run id = the - * token), which quartz's store can't serve via `#d` on a regular kind — so decisions are fetched **by - * author**, and every 46010 gate names its approver in a `p` tag, giving us exactly those authors. + * **Why we aggregate from the live subscription, not `LocalCache`:** the run/lifecycle/gate kinds are + * 46001-46031, which are neither regular (`< 10_000`) nor addressable, so `LocalCache.filter` cannot + * serve them — its note branch only matches `kind.isRegular()` events. Reading them back from the + * cache always returned nothing, so the board never showed a run. Instead we consume the events + * straight off [subscribeAsFlow], which accumulates the channel's stored + live events (deduped by id) + * and re-emits the growing list — exactly the data the aggregator needs. + * + * Two subscription realities (both mirrored from `amy buzz workflow`): the trigger + lifecycle + gate + * + the addressable definitions (30620) are `#h`-scoped to the channel — one subscription, one `#h` + * value, so Buzz's relay keeps it channel-scoped and delivers live (a multi-`#h` filter would be + * forced global and go deaf). The grant/deny decisions (46030/46031) carry only a `d` tag, so they + * ride a second subscription filtered by **author** — every 46010 gate names its approver in a `p` + * tag, giving us exactly those authors. The two streams are merged and folded on every emission. */ class WorkflowRunBoardViewModel : ViewModel() { @Volatile private var account: Account? = null @@ -98,7 +110,6 @@ class WorkflowRunBoardViewModel : ViewModel() { val isLoading: StateFlow = _isLoading.asStateFlow() private var watchJob: Job? = null - private val reloadMutex = Mutex() fun bind( account: Account, @@ -109,52 +120,60 @@ class WorkflowRunBoardViewModel : ViewModel() { this.account = account this.channelId = channelId this.relay = RelayUrlNormalizer.normalizeOrNull(relayUrl) - refresh() } - fun refresh() { - val account = account ?: return - val relay = relay ?: return - val channelId = channelId ?: return - viewModelScope.launch(Dispatchers.IO) { - _isLoading.value = true - try { - // Phase 1: the `#h`-scoped trigger + lifecycle + approval gate, plus the channel's - // addressable workflow definitions (30620) that back the picker. - account.client.fetchAllWithHooks( - filters = mapOf(relay to listOf(Filter(kinds = WORKFLOW_H_KINDS + WorkflowDefEvent.KIND, tags = mapOf("h" to listOf(channelId))))), - timeoutMs = 8_000, - pendingOnAuthRequired = true, - ) { _, _ -> false } - // Phase 2: each gate's approver can sign a decision — fetch those by author. - val approvers = approversInCache(channelId) - if (approvers.isNotEmpty()) { - account.client.fetchAllWithHooks( - filters = mapOf(relay to listOf(Filter(kinds = DECISION_KINDS, authors = approvers))), - timeoutMs = 8_000, - pendingOnAuthRequired = true, - ) { _, _ -> false } - } - reloadFromCache(channelId) - } finally { - _isLoading.value = false - } - } - } - - /** Keep the board live while on screen: any `#h` lifecycle batch (incl. 46005 completion) re-derives it. */ + /** + * Open the channel's workflow subscriptions and fold every emission into the board. The base + * `#h` subscription backfills stored runs then streams live ones; a nested by-author subscription + * (rebuilt only when the approver set actually changes) streams the grant/deny decisions. + */ + @OptIn(ExperimentalCoroutinesApi::class) fun startWatching() { val account = account ?: return val relay = relay ?: return val channelId = channelId ?: return if (watchJob != null) return + watchJob = viewModelScope.launch(Dispatchers.IO) { - account.client - .subscribeAsFlow(relay, listOf(Filter(kinds = WORKFLOW_H_KINDS + WorkflowDefEvent.KIND, tags = mapOf("h" to listOf(channelId))))) - .collect { - // The client's global listener already consumed the batch into LocalCache. - reloadFromCache(channelId) + _isLoading.value = true + // Fall out of the loading state even if the channel is genuinely empty (no EOSE signal). + val loadingTimeout = + launch { + delay(LOADING_TIMEOUT_MS) + _isLoading.value = false + } + + val baseFilter = Filter(kinds = WORKFLOW_H_KINDS + WorkflowDefEvent.KIND, tags = mapOf("h" to listOf(channelId))) + // `onStart(emptyList)` so the merged flow produces a first value even before any event + // arrives — otherwise `combine` would never emit for an empty channel and the board + // would sit on the spinner forever. + val baseFlow = account.client.subscribeAsFlow(relay, listOf(baseFilter)).onStart { emit(emptyList()) } + + val decisionFlow = + baseFlow + .map { base -> + base + .filterIsInstance() + .mapNotNull { it.approver() } + .distinct() + .sorted() + }.distinctUntilChanged() + .flatMapLatest { approvers -> + if (approvers.isEmpty()) { + flowOf(emptyList()) + } else { + account.client.subscribeAsFlow(relay, listOf(Filter(kinds = DECISION_KINDS, authors = approvers))).onStart { emit(emptyList()) } + } + } + + combine(baseFlow, decisionFlow) { base, decisions -> base + decisions } + .collect { events -> + if (events.isNotEmpty()) { + loadingTimeout.cancel() + _isLoading.value = false + } + derive(events) } } } @@ -164,85 +183,76 @@ class WorkflowRunBoardViewModel : ViewModel() { watchJob = null } - private suspend fun reloadFromCache(channelId: String) = - reloadMutex.withLock { - _definitions.value = - LocalCache - .filter(Filter(kinds = listOf(WorkflowDefEvent.KIND), tags = mapOf("h" to listOf(channelId)))) - .mapNotNull { it.event as? WorkflowDefEvent } - .map { WorkflowDefOption(it.workflowId(), it.name()?.takeIf { n -> n.isNotBlank() }, it.yaml()) } - .distinctBy { it.id } - .sortedBy { (it.name ?: it.id).lowercase() } - val base = - LocalCache - .filter(Filter(kinds = WORKFLOW_H_KINDS, tags = mapOf("h" to listOf(channelId)))) - .mapNotNull { it.event } - val approvers = - base - .filterIsInstance() - .mapNotNull { it.approver() } - .distinct() - val decisions = - if (approvers.isEmpty()) { - emptyList() - } else { - LocalCache.filter(Filter(kinds = DECISION_KINDS, authors = approvers)).mapNotNull { it.event } - } - _runs.value = WorkflowRunAggregator.byPriority(WorkflowRunAggregator.aggregate(base + decisions)) - } - - /** The approver pubkeys named by 46010 gates in this channel — who can sign a 46030/46031. */ - private fun approversInCache(channelId: String): List = - LocalCache - .filter(Filter(kinds = listOf(WorkflowApprovalRequestedEvent.KIND), tags = mapOf("h" to listOf(channelId)))) - .mapNotNull { it.event } - .filterIsInstance() - .mapNotNull { it.approver() } - .distinct() + /** Fold the merged event list into the definitions picker + the prioritized run board. */ + private fun derive(events: List) { + _definitions.value = + events + .filterIsInstance() + .map { WorkflowDefOption(it.workflowId(), it.name()?.takeIf { n -> n.isNotBlank() }, it.yaml()) } + .distinctBy { it.id } + .sortedBy { (it.name ?: it.id).lowercase() } + _runs.value = WorkflowRunAggregator.byPriority(WorkflowRunAggregator.aggregate(events)) + } fun trigger( workflowId: String, task: String, - ) = act { account, relay, channelId -> - account.triggerBuzzWorkflow(relay, channelId, workflowId, task) + onResult: (Boolean) -> Unit, + ) = act(onResult) { account, relay, channelId -> + account.triggerBuzzWorkflow(relay, channelId, workflowId, task) != null + } + + fun approve( + runId: HexKey, + onResult: (Boolean) -> Unit, + ) = act(onResult) { account, relay, _ -> + account.approveBuzzWorkflowRun(relay, runId) != null + } + + fun deny( + runId: HexKey, + onResult: (Boolean) -> Unit, + ) = act(onResult) { account, relay, _ -> + account.denyBuzzWorkflowRun(relay, runId) != null } /** * Publish a new workflow definition (kind-30620) for this channel and hand its freshly-minted id - * back on [onCreated] (on the main-relevant flow) so the picker can select it immediately. + * back on [onResult] (on the main thread) — or `null` if the account can't write / the publish + * failed, so the caller can surface an error instead of hanging the editor open silently. */ fun defineWorkflow( name: String, yaml: String, - onCreated: (String) -> Unit, + onResult: (String?) -> Unit, ) { - val account = account ?: return - val relay = relay ?: return - val channelId = channelId ?: return + val account = account + val relay = relay + val channelId = channelId + if (account == null || relay == null || channelId == null) { + onResult(null) + return + } viewModelScope.launch(Dispatchers.IO) { val newId = account.publishBuzzWorkflowDef(relay, channelId, name, yaml) - reloadFromCache(channelId) - if (newId != null) onCreated(newId) + withContext(Dispatchers.Main) { onResult(newId) } } } - fun approve(runId: HexKey) = - act { account, relay, _ -> - account.approveBuzzWorkflowRun(relay, runId) + private inline fun act( + crossinline onResult: (Boolean) -> Unit, + crossinline block: suspend (Account, NormalizedRelayUrl, String) -> Boolean, + ) { + val account = account + val relay = relay + val channelId = channelId + if (account == null || relay == null || channelId == null) { + onResult(false) + return } - - fun deny(runId: HexKey) = - act { account, relay, _ -> - account.denyBuzzWorkflowRun(relay, runId) - } - - private inline fun act(crossinline block: suspend (Account, NormalizedRelayUrl, String) -> Unit) { - val account = account ?: return - val relay = relay ?: return - val channelId = channelId ?: return viewModelScope.launch(Dispatchers.IO) { - block(account, relay, channelId) - reloadFromCache(channelId) // optimistic local re-derive; the live watch catches relay echoes + val ok = block(account, relay, channelId) + withContext(Dispatchers.Main) { onResult(ok) } } } @@ -252,22 +262,23 @@ class WorkflowRunBoardViewModel : ViewModel() { } companion object { - // The `#h`-scoped workflow kinds: trigger (46020), run/step lifecycle (46001-46007 minus the - // deny/grant commands), and the relay-signed approval mirror (46010-46012). The client-signed - // grant/deny commands (46030/46031) carry no `h` tag — fetched by author instead. + private const val LOADING_TIMEOUT_MS = 6_000L + + // The `#h`-scoped workflow kinds the aggregator can actually fold: trigger (46020), run/step + // lifecycle (46001-46003, 46005-46007) and the relay-signed approval-requested gate (46010). + // Deliberately excluded: the client-signed grant/deny commands (46030/46031) carry no `h` tag + // (fetched by author instead), and 46004/46011/46012 carry no correlatable run id, so the + // aggregator ignores them — fetching them would be pure cost. private val WORKFLOW_H_KINDS = listOf( WorkflowTriggerEvent.KIND, WorkflowTriggeredEvent.KIND, WorkflowStepStartedEvent.KIND, WorkflowStepCompletedEvent.KIND, - WorkflowStepFailedEvent.KIND, WorkflowCompletedEvent.KIND, WorkflowFailedEvent.KIND, WorkflowCancelledEvent.KIND, WorkflowApprovalRequestedEvent.KIND, - WorkflowApprovalGrantedEvent.KIND, - WorkflowApprovalDeniedEvent.KIND, ) private val DECISION_KINDS = listOf(ApprovalGrantEvent.KIND, ApprovalDenyEvent.KIND) } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt index 4d789a407b..6d06f2cc0b 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt @@ -174,7 +174,9 @@ object WorkflowRunAggregator { (thread.filterIsInstance() + thread.filterIsInstance()) .sortedBy { it.createdAt } + // Parse each event's JSON content at most once — `fold` runs per run, per board re-derive. val triggerPayload = trigger?.let { payload(it.content) } + val triggeredPayload = triggered?.let { payload(it.content) } // Terminal outcomes and the deny decision compete by timestamp — newest wins. val terminal = @@ -197,9 +199,9 @@ object WorkflowRunAggregator { return WorkflowRun( runId = runId, - workflowId = trigger?.workflowId() ?: triggerPayload?.workflow ?: payload(triggered?.content ?: "")?.workflow, + workflowId = trigger?.workflowId() ?: triggerPayload?.workflow ?: triggeredPayload?.workflow, channel = trigger?.tags?.workflowChannel() ?: approval?.channel() ?: triggered?.channel(), - task = triggerPayload?.task ?: payload(triggered?.content ?: "")?.task, + task = triggerPayload?.task ?: triggeredPayload?.task, requester = trigger?.pubKey, state = state, pendingApprover = approval?.approver(),