laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(buzz): make the workflow + job boards actually show runs (aggregate from the live subscription)

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011mApqAbr8vkLC7gUDjavu6
This commit is contained in:
Claude
2026-07-27 00:03:33 +00:00
parent aaf7affedc
commit 412dc53e38
4 changed files with 251 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/JobBoardViewModel.kt
+31 -49
View File
@@ -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<Boolean> = _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(
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardScreen.kt
+91 -36
View File
@@ -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<WorkflowRun>,
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<WorkflowDefOption>,
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<String?>(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")
}
}
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/WorkflowRunBoardViewModel.kt
+125 -114
View File
@@ -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<Boolean> = _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<WorkflowApprovalRequestedEvent>()
.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<WorkflowApprovalRequestedEvent>()
.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<HexKey> =
LocalCache
.filter(Filter(kinds = listOf(WorkflowApprovalRequestedEvent.KIND), tags = mapOf("h" to listOf(channelId))))
.mapNotNull { it.event }
.filterIsInstance<WorkflowApprovalRequestedEvent>()
.mapNotNull { it.approver() }
.distinct()
/** Fold the merged event list into the definitions picker + the prioritized run board. */
private fun derive(events: List<Event>) {
_definitions.value =
events
.filterIsInstance<WorkflowDefEvent>()
.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)
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt
+4 -2
View File
@@ -174,7 +174,9 @@ object WorkflowRunAggregator {
(thread.filterIsInstance<WorkflowStepStartedEvent>() + thread.filterIsInstance<WorkflowStepCompletedEvent>())
.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(),