feat(workflows): add dynamic declarative workflows

.github/workflows/pr-build.yml

@@ -125,6 +125,8 @@ jobs:
           packages/ui/src/stores/session-metadata.test.ts
           packages/ui/src/stores/session-pagination.test.ts
           packages/ui/src/stores/workspace-list-reconciliation-fence.test.ts
+          packages/ui/src/stores/workflows.test.ts
+          packages/opencode-plugin/plugin/lib/workflows.test.ts
 
       - name: Test restore ownership integration
         run: >-

packages/electron-app/electron/main/client-state-cross-host.test.ts

@@ -9,6 +9,7 @@ import test from "node:test"
 import {
   CrossHostRegistration,
   CROSS_HOST_OWNER_DIRECTORY,
+  crossHostParticipants,
   resolveCrossHostElectionDirectory,
   resolveCrossHostStatePath,
   type CrossHostLeaseDependencies,
@@ -141,6 +142,28 @@ test("simultaneous claimants deterministically recover a stale owner", (t) => {
   assert.equal(loser.isPrimary, false)
 })
 
+test("async recovery replaces a prior claim for a consecutive crashed owner", async (t) => {
+  const directory = temp(t), candidate = owner(620, "candidate")
+  const registration = CrossHostRegistration.register(directory, candidate, false, {
+    pidAlive: () => true,
+    processStartIdentity: () => { throw new Error("sync identity lookup must not run") },
+    processStartIdentityAsync: async () => "reused-start",
+  })!
+  const publishStaleOwner = (stale: ProcessOwner) => {
+    mkdirSync(join(directory, CROSS_HOST_OWNER_DIRECTORY))
+    writeFileSync(ownerFile(directory), JSON.stringify(stale))
+  }
+  const first = owner(621, "first", "first-start")
+  publishStaleOwner(first)
+  assert.equal(await registration.tryAcquireAsync(true), true)
+  rmSync(join(directory, CROSS_HOST_OWNER_DIRECTORY), { recursive: true, force: true })
+
+  const second = owner(622, "second", "second-start")
+  publishStaleOwner(second)
+  assert.equal(await registration.tryAcquireAsync(true), true)
+  assert.equal(readFileSync(join(directory, "recovery.620.candidate.claim"), "utf8"), JSON.stringify(second))
+})
+
 test("graceful primary release allows a successor while a secondary remains", (t) => {
   const directory = temp(t)
   const primary = CrossHostRegistration.register(directory, owner(401, "primary"), true, dependencies(true, "primary-start"))!
@@ -171,11 +194,15 @@ test("graceful handoff retires the old cohort so a crashed successor can recover
   CrossHostRegistration.register(directory, secondaryOwner, true, dependencies(true, "primary-start"))!
 
   primary.release()
-  assert.equal(readdirSync(directory).some((name) => name.startsWith("retired.")), false)
+  assert.equal(readdirSync(directory).some((name) => name.startsWith("retired.participant.")), true)
   assert.equal(JSON.parse(readFileSync(ownerFile(directory), "utf8")).runToken, "successor")
   assert.equal(existsSync(join(directory, "participant.423.successor.json")), false)
   assert.equal(existsSync(join(directory, "participant.425.late.json")), false)
   assert.equal(existsSync(malformed), false)
+  assert.deepEqual(
+    crossHostParticipants(directory).map(({ runToken }) => runToken).sort(),
+    ["late", "secondary", "successor"],
+  )
 
   const claimantOwner = owner(424, "claimant"), identities = new Map([
     [secondaryOwner.pid, secondaryOwner.processStartIdentity],

packages/electron-app/electron/main/client-state-cross-host.ts

@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"
 import { closeSync, existsSync, fsyncSync, linkSync, mkdirSync, openSync, readdirSync, readFileSync, renameSync, rmSync, unlinkSync, writeFileSync } from "node:fs"
 import { homedir } from "node:os"
 import { basename, dirname, join, posix, win32 } from "node:path"
-import { getProcessStartIdentity, type ProcessStartIdentityLookup } from "./client-state-process-identity"
+import { getProcessStartIdentity, getProcessStartIdentityAsync, type AsyncProcessStartIdentityLookup, type ProcessStartIdentityLookup } from "./client-state-process-identity"
 import { hasErrorCode, isPidAlive, type ProcessOwner } from "./client-state-process"
 
 export const CROSS_HOST_OWNER_DIRECTORY = "primary.owner.json"
@@ -12,11 +12,14 @@ const PARTICIPANT_SUFFIX = ".json"
 const RECOVERY_PREFIX = "recovery."
 const RECOVERY_SUFFIX = ".claim"
 const RETIRED_PREFIX = "retired."
+const RETIRED_PARTICIPANT_PREFIX = "retired.participant."
+const RETIRED_PARTICIPANT_SUFFIX = ".json"
 const ACQUIRE_ATTEMPTS = 10
 
 export interface CrossHostLeaseDependencies {
   pidAlive(pid: number): boolean
   processStartIdentity: ProcessStartIdentityLookup
+  processStartIdentityAsync?: AsyncProcessStartIdentityLookup
   onParticipantPublished?(): void
   onOwnerPrepared?(): void
   onOwnerRetired?(): void
@@ -26,6 +29,7 @@ export interface CrossHostLeaseDependencies {
 const defaultDependencies: CrossHostLeaseDependencies = {
   pidAlive: isPidAlive,
   processStartIdentity: getProcessStartIdentity,
+  processStartIdentityAsync: getProcessStartIdentityAsync,
 }
 
 function validHome(value: string | undefined, platform: NodeJS.Platform): string | undefined {
@@ -137,13 +141,28 @@ function recoveryPath(directory: string, owner: ProcessOwner): string {
   return join(directory, `${RECOVERY_PREFIX}${owner.pid}.${owner.runToken}${RECOVERY_SUFFIX}`)
 }
 
+function retiredParticipantPath(directory: string, owner: ProcessOwner): string {
+  return join(directory, `${RETIRED_PARTICIPANT_PREFIX}${owner.pid}.${owner.runToken}${RETIRED_PARTICIPANT_SUFFIX}`)
+}
+
 function publishParticipant(path: string, owner: ProcessOwner): void {
   const value = serializeOwner(owner)
   try { publishFile(path, value) } catch (error) {
     if (!hasErrorCode(error, "EEXIST") || readIfExists(path) !== value) throw error
   }
 }
 
+function publishRecoveryClaim(path: string, observedOwner: string): void {
+  if (readIfExists(path) !== observedOwner) {
+    try { unlinkSync(path) } catch (error) {
+      if (!hasErrorCode(error, "ENOENT")) throw error
+    }
+  }
+  try { publishFile(path, observedOwner) } catch (error) {
+    if (!hasErrorCode(error, "EEXIST") || readIfExists(path) !== observedOwner) throw error
+  }
+}
+
 function ownerPath(directory: string): string {
   return join(directory, CROSS_HOST_OWNER_DIRECTORY, OWNER_FILENAME)
 }
@@ -154,6 +173,14 @@ function ownerIsStale(owner: ProcessOwner, dependencies: CrossHostLeaseDependenc
   return identity ? identity !== owner.processStartIdentity : undefined
 }
 
+async function ownerIsStaleAsync(owner: ProcessOwner, dependencies: CrossHostLeaseDependencies): Promise<boolean | undefined> {
+  if (!dependencies.pidAlive(owner.pid)) return true
+  const identity = dependencies.processStartIdentityAsync
+    ? await dependencies.processStartIdentityAsync(owner.pid, 1_000)
+    : dependencies.processStartIdentity(owner.pid)
+  return identity ? identity !== owner.processStartIdentity : undefined
+}
+
 function removeParticipantIfOwned(path: string, owner: ProcessOwner): void {
   const observed = readIfExists(path)
   const current = observed === undefined ? undefined : parseOwner(observed)
@@ -163,6 +190,10 @@ function removeParticipantIfOwned(path: string, owner: ProcessOwner): void {
   }
 }
 
+function removeRetiredParticipantIfOwned(directory: string, owner: ProcessOwner): void {
+  removeParticipantIfOwned(retiredParticipantPath(directory, owner), owner)
+}
+
 function retireOwnerIfOwned(directory: string, owner: ProcessOwner, dependencies: CrossHostLeaseDependencies): void {
   const observed = readIfExists(ownerPath(directory))
   const current = parseOwner(observed ?? "")
@@ -182,6 +213,7 @@ function retireOwnerIfOwned(directory: string, owner: ProcessOwner, dependencies
       if (observedParticipant === undefined) continue
       const participant = parseOwner(observedParticipant)
       if (participant) {
+        publishParticipant(retiredParticipantPath(directory, participant), participant)
         removeParticipantIfOwned(path, participant)
         try { unlinkSync(recoveryPath(directory, participant)) } catch {}
       } else if (readIfExists(path) === observedParticipant) {
@@ -226,6 +258,36 @@ function recoveryClaimants(
   return claimants
 }
 
+async function recoveryClaimantsAsync(
+  directory: string,
+  current: ProcessOwner,
+  observedOwner: string,
+  dependencies: CrossHostLeaseDependencies,
+): Promise<ProcessOwner[] | undefined> {
+  const claimants = [current]
+  for (const name of readdirSync(directory)) {
+    if (!name.startsWith(PARTICIPANT_PREFIX) || !name.endsWith(PARTICIPANT_SUFFIX)) continue
+    const path = join(directory, name)
+    const participant = parseOwner(readIfExists(path) ?? "")
+    if (!participant) return undefined
+    if (sameOwner(participant, current)) continue
+    if (await ownerIsStaleAsync(participant, dependencies) === true) {
+      removeParticipantIfOwned(path, participant)
+      try { unlinkSync(recoveryPath(directory, participant)) } catch {}
+      continue
+    }
+    const claimPath = recoveryPath(directory, participant)
+    let claim = readIfExists(claimPath)
+    for (let attempt = 0; claim !== observedOwner && attempt < 20; attempt += 1) {
+      await new Promise((resolve) => setTimeout(resolve, 5))
+      claim = readIfExists(claimPath)
+    }
+    if (claim !== observedOwner) return undefined
+    claimants.push(participant)
+  }
+  return claimants
+}
+
 function retireOwner(directory: string, observed: string, owner: ProcessOwner, claimant: ProcessOwner, dependencies: CrossHostLeaseDependencies): boolean {
   if (ownerIsStale(owner, dependencies) !== true) return false
   const claimants = recoveryClaimants(directory, claimant, observed, dependencies)
@@ -244,6 +306,24 @@ function retireOwner(directory: string, observed: string, owner: ProcessOwner, c
   }
 }
 
+async function retireOwnerAsync(directory: string, observed: string, owner: ProcessOwner, claimant: ProcessOwner, dependencies: CrossHostLeaseDependencies): Promise<boolean> {
+  if (await ownerIsStaleAsync(owner, dependencies) !== true) return false
+  const claimants = await recoveryClaimantsAsync(directory, claimant, observed, dependencies)
+  if (!claimants) return false
+  claimants.sort((left, right) => serializeOwner(left) < serializeOwner(right) ? -1 : 1)
+  if (!sameOwner(claimants[0]!, claimant)) return false
+  if (readIfExists(ownerPath(directory)) !== observed) return false
+  const retired = join(directory, `${RETIRED_PREFIX}${owner.pid}.${owner.runToken}`)
+  try {
+    renameSync(join(directory, CROSS_HOST_OWNER_DIRECTORY), retired)
+    dependencies.onOwnerRetired?.()
+    return true
+  } catch (error) {
+    if (["ENOENT", "EEXIST", "ENOTEMPTY"].some((code) => hasErrorCode(error, code)) || existsSync(retired)) return false
+    throw error
+  }
+}
+
 function publishOwner(directory: string, owner: ProcessOwner, dependencies: CrossHostLeaseDependencies): boolean {
   const temporary = join(directory, `.owner.${randomUUID()}.tmp`)
   try {
@@ -264,7 +344,10 @@ function publishOwner(directory: string, owner: ProcessOwner, dependencies: Cros
 export function crossHostParticipants(directory: string): ProcessOwner[] {
   try {
     return readdirSync(directory)
-      .filter((name) => name.startsWith(PARTICIPANT_PREFIX) && name.endsWith(PARTICIPANT_SUFFIX))
+      .filter((name) =>
+        (name.startsWith(PARTICIPANT_PREFIX) && name.endsWith(PARTICIPANT_SUFFIX)) ||
+        (name.startsWith(RETIRED_PARTICIPANT_PREFIX) && name.endsWith(RETIRED_PARTICIPANT_SUFFIX)),
+      )
       .map((name) => parseOwner(readIfExists(join(directory, name)) ?? ""))
       .filter((owner): owner is ProcessOwner => Boolean(owner))
   } catch (error) {
@@ -280,7 +363,7 @@ export class CrossHostRegistration {
     private readonly directory: string,
     readonly owner: ProcessOwner,
     private readonly participant: string,
-    private readonly recoveryClaim: string | undefined,
+    private recoveryClaim: string | undefined,
     private primary: boolean,
     private readonly dependencies: CrossHostLeaseDependencies,
   ) {}
@@ -297,6 +380,7 @@ export class CrossHostRegistration {
     mkdirSync(directory, { recursive: true, mode: 0o700 })
     const participant = participantPath(directory, owner)
     publishParticipant(participant, owner)
+    removeRetiredParticipantIfOwned(directory, owner)
     dependencies.onParticipantPublished?.()
     let primary = false
     let recoveryClaim: string | undefined
@@ -311,9 +395,7 @@ export class CrossHostRegistration {
           if (sameOwner(existing, owner)) { primary = true; break }
           if (ownerIsStale(existing, dependencies) === true) {
             recoveryClaim ??= recoveryPath(directory, owner)
-            try { publishFile(recoveryClaim, observed) } catch (error) {
-              if (!hasErrorCode(error, "EEXIST") || readIfExists(recoveryClaim) !== observed) throw error
-            }
+            publishRecoveryClaim(recoveryClaim, observed)
           }
           if (!retireOwner(directory, observed, existing, owner, dependencies)) break
         }
@@ -332,10 +414,59 @@ export class CrossHostRegistration {
     return Boolean(current && sameOwner(current, this.owner))
   }
 
+  tryAcquire(primaryCandidate: boolean | (() => boolean)): boolean {
+    if (this.released || this.isPrimary) return this.isPrimary
+    if (!(typeof primaryCandidate === "function" ? primaryCandidate() : primaryCandidate)) return false
+    publishParticipant(this.participant, this.owner)
+    removeRetiredParticipantIfOwned(this.directory, this.owner)
+    for (let attempt = 0; attempt < ACQUIRE_ATTEMPTS; attempt += 1) {
+      if (publishOwner(this.directory, this.owner, this.dependencies)) { this.primary = true; break }
+      const observed = readIfExists(ownerPath(this.directory))
+      if (observed === undefined) continue
+      const existing = parseOwner(observed)
+      if (!existing) break
+      if (sameOwner(existing, this.owner)) { this.primary = true; break }
+      if (ownerIsStale(existing, this.dependencies) === true) {
+        this.recoveryClaim ??= recoveryPath(this.directory, this.owner)
+        publishRecoveryClaim(this.recoveryClaim, observed)
+      }
+      if (!retireOwner(this.directory, observed, existing, this.owner, this.dependencies)) break
+    }
+    return this.isPrimary
+  }
+
+  async tryAcquireAsync(primaryCandidate: boolean | (() => boolean)): Promise<boolean> {
+    if (this.released || this.isPrimary) return this.isPrimary
+    if (!(typeof primaryCandidate === "function" ? primaryCandidate() : primaryCandidate)) return false
+    publishParticipant(this.participant, this.owner)
+    removeRetiredParticipantIfOwned(this.directory, this.owner)
+    for (let attempt = 0; attempt < ACQUIRE_ATTEMPTS; attempt += 1) {
+      if (publishOwner(this.directory, this.owner, this.dependencies)) { this.primary = true; break }
+      const observed = readIfExists(ownerPath(this.directory))
+      if (observed === undefined) continue
+      const existing = parseOwner(observed)
+      if (!existing) break
+      if (sameOwner(existing, this.owner)) { this.primary = true; break }
+      if (await ownerIsStaleAsync(existing, this.dependencies) === true) {
+        this.recoveryClaim ??= recoveryPath(this.directory, this.owner)
+        publishRecoveryClaim(this.recoveryClaim, observed)
+      }
+      if (!await retireOwnerAsync(this.directory, observed, existing, this.owner, this.dependencies)) break
+    }
+    return this.isPrimary
+  }
+
+  deferPrimary(): void {
+    if (this.released) return
+    retireOwnerIfOwned(this.directory, this.owner, this.dependencies)
+    this.primary = false
+  }
+
   release(): boolean {
     if (this.released) return false
     retireOwnerIfOwned(this.directory, this.owner, this.dependencies)
     removeParticipantIfOwned(this.participant, this.owner)
+    removeRetiredParticipantIfOwned(this.directory, this.owner)
     if (this.recoveryClaim) try { unlinkSync(this.recoveryClaim) } catch {}
     this.primary = false
     this.released = true

packages/electron-app/electron/main/client-state-process-identity.ts

@@ -1,11 +1,12 @@
 import { execFile, spawnSync } from "node:child_process"
-import { readFile as readFileAsync } from "node:fs/promises"
+import { readFile as readFileAsync, readlink as readlinkAsync } from "node:fs/promises"
 import { readFileSync, readlinkSync } from "node:fs"
 import { basename, resolve } from "node:path"
 
 export type ProcessStartIdentityLookup = (pid: number) => string | undefined
 export type AsyncProcessStartIdentityLookup = (pid: number, timeoutMs: number) => Promise<string | undefined> | string | undefined
 export type ExpectedProcessLookup = (pid: number) => boolean | undefined
+export type AsyncExpectedProcessLookup = (pid: number, timeoutMs: number) => Promise<boolean | undefined> | boolean | undefined
 
 function readLinuxProcessStartIdentity(pid: number): string | undefined {
   const stat = readFileSync(`/proc/${pid}/stat`, "utf8")
@@ -127,3 +128,29 @@ export function isExpectedTauriProcess(pid: number): boolean | undefined {
     return undefined
   }
 }
+
+export async function isExpectedTauriProcessAsync(
+  pid: number,
+  timeoutMs: number,
+  platform: NodeJS.Platform = process.platform,
+): Promise<boolean | undefined> {
+  if (!Number.isInteger(pid) || pid <= 0 || timeoutMs <= 0) return undefined
+  try {
+    const executable = platform === "linux"
+      ? await readlinkAsync(`/proc/${pid}/exe`)
+      : await readCommandIdentityAsync(
+          platform === "win32" ? "powershell.exe" : "ps",
+          platform === "win32"
+            ? ["-NoProfile", "-NonInteractive", "-Command", `(Get-Process -Id ${pid} -ErrorAction Stop).Path`]
+            : ["-p", String(pid), "-o", "comm="],
+          "path",
+          timeoutMs,
+        ).then((value) => value?.slice(5))
+    if (!executable) return undefined
+    if (resolve(executable).toLowerCase() === resolve(process.execPath).toLowerCase()) return false
+    return ["codenomad", "codenomad.exe", "codenomad-tauri", "codenomad-tauri.exe"]
+      .includes(basename(executable).toLowerCase())
+  } catch {
+    return undefined
+  }
+}

packages/electron-app/electron/main/client-state-process.test.ts

@@ -165,6 +165,29 @@ test("a surviving older secondary keeps later processes secondary", async (t) =>
   assert.equal(third.role.isPrimary, false)
 })
 
+test("a retained secondary retries after the local primary exits", (t) => {
+  const directory = temp(t)
+  const primaryLockPath = join(directory, "client-state.primary.lock")
+  const registrationLockPath = join(directory, "client-state.registration.lock")
+  const paths = { primaryLockPath, registrationLockPath }
+  const identities = new Map([[31, "first-start"], [32, "second-start"], [33, "third-start"]])
+  const alive = (pid: number) => identities.has(pid)
+  const identity = (pid: number) => identities.get(pid)
+  const first = { pid: 31, runToken: "first", processStartIdentity: "first-start" }
+  const second = { pid: 32, runToken: "second", processStartIdentity: "second-start" }
+  const third = { pid: 33, runToken: "third", processStartIdentity: "third-start" }
+
+  assert.equal(electClientStateProcess(directory, first, paths, () => {}, alive, 0, () => {}, identity), true)
+  assert.equal(electClientStateProcess(directory, second, paths, () => {}, alive, 0, () => {}, identity), false)
+  assert.equal(electClientStateProcess(directory, third, paths, () => {}, alive, 0, () => {}, identity), false)
+  removeProcessOwnerLockIfOwned(primaryLockPath, first)
+  removeRunningMarkerIfOwned(getRunningMarkerPath(directory, first), first)
+  identities.delete(first.pid)
+
+  assert.equal(electClientStateProcess(directory, second, paths, () => {}, alive, 0, () => {}, identity, true), true)
+  assert.equal(electClientStateProcess(directory, third, paths, () => {}, alive, 0, () => {}, identity, true), false)
+})
+
 test("lock recovery handles PID reuse, malformed files, and verified live owners", async (t) => {
   const cases = [
     { name: "same PID old token", owner: { pid: process.pid, runToken: "new" }, file: { pid: process.pid, runToken: "old" }, lock: "primary", alive: () => true, expected: true },