From 2adc42a035e825af7370669a6e03fc936a298d5f Mon Sep 17 00:00:00 2001
From: pescao0 <269974397+pescao0@users.noreply.github.com>
Date: Sun, 30 Aug 2026 09:35:45 +0200
Subject: [PATCH] feat(tui): add prompt transform coordinator

---
 packages/tui/src/prompt/transform.ts       | 227 +++++++++++++++++++++
 packages/tui/test/prompt/transform.test.ts | 173 ++++++++++++++++
 2 files changed, 400 insertions(+)
 create mode 100644 packages/tui/src/prompt/transform.ts
 create mode 100644 packages/tui/test/prompt/transform.test.ts

diff --git a/packages/tui/src/prompt/transform.ts b/packages/tui/src/prompt/transform.ts
new file mode 100644
index 000000000000..c79dcbbc33f9
--- /dev/null
+++ b/packages/tui/src/prompt/transform.ts
@@ -0,0 +1,227 @@
+const RESERVED_CONTEXT_KEYS = ["draftID", "revision", "signal"] as const
+type ReservedContextKey = (typeof RESERVED_CONTEXT_KEYS)[number]
+export type DeepReadonly<Value> = Value extends readonly (infer Item)[]
+  ? readonly DeepReadonly<Item>[]
+  : Value extends object
+    ? { readonly [Key in keyof Value]: DeepReadonly<Value[Key]> }
+    : Value
+export type PromptTransformCurrent = { revision: number; fingerprint: string }
+export type PromptTransformCallerContext<Context extends object> = Omit<Context, ReservedContextKey> &
+  Partial<Record<ReservedContextKey, never>>
+export type PromptTransformContext<Context extends object> = DeepReadonly<Omit<Context, ReservedContextKey>> & {
+  readonly draftID: string
+  readonly revision: number
+  readonly signal: AbortSignal
+}
+export type PromptTransformRegistration<Draft extends object, Context extends object> = {
+  id: string
+  transform: (draft: DeepReadonly<Draft>, context: PromptTransformContext<Context>) => Draft | Promise<Draft>
+}
+export type PromptTransformRequest<Draft extends object, Context extends object> = PromptTransformCurrent & {
+  draftID: string
+  draft: Draft
+  context: PromptTransformCallerContext<Context>
+  current: () => PromptTransformCurrent
+  commit: (draft: Draft) => undefined
+}
+type Registration<Draft extends object, Context extends object> = PromptTransformRegistration<Draft, Context> & {
+  jobs: Set<Job<Draft, Context>>
+}
+type Accepted<Draft extends object, Context extends object> = Omit<
+  PromptTransformRequest<Draft, Context>,
+  "draft" | "context"
+> & {
+  draft: DeepReadonly<Draft>
+  context: DeepReadonly<PromptTransformCallerContext<Context>>
+}
+type Job<Draft extends object, Context extends object> = Accepted<Draft, Context> & {
+  steps: Registration<Draft, Context>[]
+  state: DraftState<Draft, Context>
+  controller: AbortController
+  settled: PromiseWithResolvers<boolean>
+  dead: boolean
+  commitStarted: boolean
+}
+type DraftState<Draft extends object, Context extends object> = {
+  latest: PromptTransformCurrent
+  active?: Job<Draft, Context>
+  queued?: Job<Draft, Context>
+}
+export function createPromptTransformCoordinator<Draft extends object, Context extends object>(
+  fingerprint: (draft: DeepReadonly<Draft>) => string,
+) {
+  const registrations = new Map<string, Registration<Draft, Context>>()
+  const drafts = new Map<string, DraftState<Draft, Context>>()
+  const disposedDrafts = new Set<string>()
+  function register(input: PromptTransformRegistration<Draft, Context>) {
+    const id = input.id
+    if (registrations.has(id)) throw new Error(`Prompt transform "${id}" is already registered`)
+    const registration: Registration<Draft, Context> = { ...input, id, jobs: new Set() }
+    // Map insertion order is transform order; deleting and re-registering appends after survivors.
+    registrations.set(id, registration)
+    let disposed = false
+    return () => {
+      if (disposed) return
+      disposed = true
+      registrations.delete(id)
+      cancel([...registration.jobs])
+    }
+  }
+  function apply(request: PromptTransformRequest<Draft, Context>): Promise<boolean> {
+    if (disposedDrafts.has(request.draftID)) return Promise.resolve(false)
+    if (RESERVED_CONTEXT_KEYS.some((key) => key in request.context)) {
+      throw new TypeError("Prompt transform context contains reserved metadata")
+    }
+    const accepted: Accepted<Draft, Context> = {
+      ...request,
+      draft: immutablePlain(request.draft),
+      context: immutablePlain(request.context),
+    }
+    if (fingerprint(accepted.draft) !== accepted.fingerprint || !matches(snapshotCurrent(accepted.current), accepted)) {
+      return Promise.resolve(false)
+    }
+    const state = drafts.get(accepted.draftID) ?? { latest: { revision: -1, fingerprint: "" } }
+    drafts.set(accepted.draftID, state)
+    if (accepted.revision < state.latest.revision) return Promise.resolve(false)
+    if (accepted.revision === state.latest.revision && accepted.fingerprint !== state.latest.fingerprint) {
+      return Promise.resolve(false)
+    }
+    if (state.active && live(state.latest, state.active) && matches(state.active, accepted))
+      return state.active.settled.promise
+    if (state.queued && !state.queued.controller.signal.aborted && matches(state.queued, accepted))
+      return state.queued.settled.promise
+    const replaced = [state.active, state.queued]
+    const job = createJob(accepted, state)
+    state.latest = accepted
+    state.queued = job
+    cancel(replaced)
+    runNext(state)
+    return job.settled.promise
+  }
+  function createJob(accepted: Accepted<Draft, Context>, state: DraftState<Draft, Context>) {
+    const snapshot = [...registrations.values()]
+    const job = accepted as Job<Draft, Context>
+    job.steps = snapshot
+    job.state = state
+    job.controller = new AbortController()
+    job.settled = Promise.withResolvers()
+    job.dead = false
+    job.commitStarted = false
+    snapshot.forEach((registration) => registration.jobs.add(job))
+    const cancel = () => {
+      detach(job)
+      if (!job.commitStarted && job.state.active !== job) job.settled.resolve(false)
+    }
+    job.controller.signal.addEventListener("abort", cancel, { once: true })
+    return job
+  }
+  function runNext(state: DraftState<Draft, Context>) {
+    if (state.active || !state.queued) return
+    const job = state.queued
+    state.queued = undefined
+    if (job.dead || job.controller.signal.aborted || !matches(state.latest, job)) {
+      job.controller.abort()
+      runNext(state)
+      return
+    }
+    state.active = job
+    let committed = false
+    const failures: unknown[] = []
+    void compose(job)
+      .then((draft) => {
+        if (draft === undefined || !live(state.latest, job)) return
+        const observed = snapshotCurrent(job.current)
+        const resultFingerprint = fingerprint(draft)
+        if (resultFingerprint === job.fingerprint || !live(state.latest, job) || !matches(observed, job)) return
+        const next = structuredClone(draft) as Draft
+        const confirmed = snapshotCurrent(job.current)
+        if (!live(state.latest, job) || !matches(confirmed, job)) return
+        job.commitStarted = true
+        job.commit(next)
+        committed = true
+      })
+      .catch((error) => {
+        if (!job.commitStarted && job.controller.signal.aborted) return
+        failures.push(error)
+      })
+      .finally(() => {
+        detach(job)
+        if (state.active === job) state.active = undefined
+        runNext(state)
+        if (failures.length) job.settled.reject(failures[0])
+        else job.settled.resolve(committed)
+      })
+  }
+  function detach(job: Job<Draft, Context>) {
+    job.steps.forEach((registration) => registration.jobs.delete(job))
+  }
+  function cancel(items: Array<Job<Draft, Context> | undefined>) {
+    items.forEach((job) => {
+      if (!job) return
+      if (job.state.queued === job) job.state.queued = undefined
+      job.dead = true
+    })
+    items.forEach((job) => job?.controller.abort())
+  }
+  function disposeDraft(draftID: string) {
+    if (disposedDrafts.has(draftID)) return
+    disposedDrafts.add(draftID)
+    const state = drafts.get(draftID)
+    if (!state) return
+    const affected = [state.active, state.queued]
+    Object.assign(state, { active: undefined, queued: undefined })
+    drafts.delete(draftID)
+    cancel(affected)
+  }
+  return { register, apply, disposeDraft }
+}
+async function compose<Draft extends object, Context extends object>(job: Job<Draft, Context>) {
+  let draft = job.draft
+  for (const registration of job.steps) {
+    if (job.controller.signal.aborted) return
+    const context = immutableContext(job.context, job.draftID, job.revision, job.controller.signal)
+    draft = immutablePlain(await registration.transform(draft, context))
+  }
+  return draft
+}
+function live<Draft extends object, Context extends object>(current: PromptTransformCurrent, job: Job<Draft, Context>) {
+  return !job.dead && !job.controller.signal.aborted && matches(current, job)
+}
+function matches(current: PromptTransformCurrent, expected: PromptTransformCurrent) {
+  return current.revision === expected.revision && current.fingerprint === expected.fingerprint
+}
+function snapshotCurrent(current: () => PromptTransformCurrent): PromptTransformCurrent {
+  const value = current()
+  return { revision: value.revision, fingerprint: value.fingerprint }
+}
+function immutableContext<Context extends object>(
+  context: DeepReadonly<PromptTransformCallerContext<Context>>,
+  draftID: string,
+  revision: number,
+  signal: AbortSignal,
+) {
+  const value = Object.assign(structuredClone(context), { draftID, revision, signal })
+  return freeze(value) as PromptTransformContext<Context>
+}
+function immutablePlain<Value extends object>(value: Value): DeepReadonly<Value> {
+  if (!value || typeof value !== "object" || Array.isArray(value))
+    throw new TypeError("Prompt transform data must be a plain object")
+  assertPlain(value, new Set())
+  return freeze(structuredClone(value)) as DeepReadonly<Value>
+}
+function assertPlain(value: unknown, seen: Set<object>) {
+  if (value === null || ["string", "number", "boolean", "bigint", "undefined"].includes(typeof value)) return
+  if (typeof value !== "object") throw new TypeError("Prompt transform data must contain only plain structured data")
+  if (seen.has(value)) return
+  if (!Array.isArray(value) && ![Object.prototype, null].includes(Object.getPrototypeOf(value))) {
+    throw new TypeError("Prompt transform data must contain only plain structured data")
+  }
+  seen.add(value)
+  Object.values(value).forEach((item) => assertPlain(item, seen))
+}
+function freeze(value: unknown, seen = new Set<object>()): unknown {
+  if (!value || typeof value !== "object" || seen.has(value)) return value
+  seen.add(value)
+  Object.values(value).forEach((item) => freeze(item, seen))
+  return Object.freeze(value)
+}
diff --git a/packages/tui/test/prompt/transform.test.ts b/packages/tui/test/prompt/transform.test.ts
new file mode 100644
index 000000000000..8773d7893024
--- /dev/null
+++ b/packages/tui/test/prompt/transform.test.ts
@@ -0,0 +1,173 @@
+import { describe, expect, test } from "bun:test"
+import { createPromptTransformCoordinator, type DeepReadonly } from "../../src/prompt/transform"
+type Draft = { text: string; detail: { values: string[] }; self?: Draft }
+type Context = { metadata: { label: string } }
+const draft = (text = "source"): Draft => ({ text, detail: { values: ["value"] } })
+const fingerprint = (value: DeepReadonly<Draft>) => JSON.stringify({ text: value.text, detail: value.detail })
+function harness(text = "source", revision = 1, draftID = "draft-1") {
+  const value = draft(text)
+  const current = { revision, fingerprint: fingerprint(value) }
+  const commits: Draft[] = []
+  const request = {
+    draftID,
+    revision,
+    draft: value,
+    fingerprint: current.fingerprint,
+    context: { metadata: { label: "original" } },
+    current: () => current,
+    commit: (next: Draft) => (commits.push(next), undefined),
+  }
+  return { current, commits, request }
+}
+function revise(run: ReturnType<typeof harness>, value: Draft, revision: number) {
+  Object.assign(run.current, { revision, fingerprint: fingerprint(value) })
+  return { ...run.request, revision, draft: value, fingerprint: run.current.fingerprint }
+}
+describe("prompt transform coordinator", () => {
+  test("orders registrations and snapshots a mutable registration ID", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    const calls: string[] = []
+    let completedAborts = 0
+    const transform = (id: string) => async (_: DeepReadonly<Draft>, context: { signal: AbortSignal }) => {
+      calls.push(id)
+      if (calls.length === 1) context.signal.addEventListener("abort", () => completedAborts++)
+      return draft(id)
+    }
+    const input = { id: "first", transform: transform("first") }
+    const dispose = coordinator.register(input)
+    coordinator.register({ id: "second", transform: transform("second") })
+    input.id = "changed"
+    dispose()
+    coordinator.register({ id: "first", transform: transform("first") })
+    const run = harness()
+    const first = coordinator.apply(run.request)
+    let replay!: Promise<boolean>
+    await first.then(() => {
+      replay = coordinator.apply(run.request)
+      coordinator.disposeDraft(run.request.draftID)
+    })
+    expect(calls).toEqual(["second", "first", "second"])
+    expect(completedAborts).toBe(0)
+  })
+  test("freezes cyclic boundaries, rejects built-ins, and detaches commit input", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    const run = harness()
+    run.request.draft.self = run.request.draft
+    let output!: Draft
+    coordinator.register({
+      id: "first",
+      transform: async (value, context) => {
+        expect(value.self).toBe(value)
+        expect([value, value.detail, context, context.metadata, context.signal].every(Object.isFrozen)).toBe(true)
+        return (output = draft(`${value.text}:first`))
+      },
+    })
+    coordinator.register({
+      id: "second",
+      transform: async (value) => {
+        expect(value).not.toBe(output)
+        expect(Object.isFrozen(value)).toBe(true)
+        return (output = draft(`${value.text}:second`))
+      },
+    })
+    expect([await coordinator.apply(run.request), run.commits[0] === output]).toEqual([true, false])
+    const invalid = { ...run.request.context, mutable: new Map() } as Context
+    expect(() => coordinator.apply({ ...run.request, context: invalid })).toThrow("plain structured data")
+  })
+  test("snapshots and coalesces active and queued requests", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    const gate = Promise.withResolvers<void>()
+    coordinator.register({
+      id: "controlled",
+      async transform(value, context) {
+        if (context.revision === 1) await gate.promise
+        return draft(`${value.text}:${context.metadata.label}`)
+      },
+    })
+    const run = harness()
+    const active = coordinator.apply(run.request)
+    expect(coordinator.apply(run.request)).toBe(active)
+    run.current.revision = 2
+    const queued = { ...run.request, revision: 2 }
+    const next = coordinator.apply(queued)
+    expect(coordinator.apply(queued)).toBe(next)
+    queued.revision = 9
+    queued.draftID = queued.fingerprint = "changed"
+    queued.current = () => ({ revision: 9, fingerprint: "changed" })
+    queued.commit = () => undefined
+    queued.draft.text = "changed"
+    queued.context.metadata.label = "changed"
+    gate.resolve()
+    expect(await Promise.all([active, next])).toEqual([false, true])
+    expect(run.commits[0]?.text).toBe("source:original")
+  })
+  test("preserves reentrant work published by a superseding abort", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    const gate = Promise.withResolvers<void>()
+    const run = harness()
+    let reentrant!: Promise<boolean>
+    coordinator.register({
+      id: "controlled",
+      async transform(value, context) {
+        if (context.revision === 1) {
+          context.signal.addEventListener("abort", () => {
+            run.current.revision = 3
+            reentrant = coordinator.apply({ ...run.request, revision: 3 })
+          })
+          await gate.promise
+        }
+        return draft(`${value.text}:${context.revision}`)
+      },
+    })
+    const active = coordinator.apply(run.request)
+    run.current.revision = 2
+    const superseded = coordinator.apply({ ...run.request, revision: 2 })
+    gate.resolve()
+    expect(await Promise.all([active, superseded, reentrant])).toEqual([false, false, true])
+  })
+  test("registration disposal publishes cancellation before abort listeners run", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    const run = harness()
+    let reentrant!: Promise<boolean>
+    const dispose = coordinator.register({
+      id: "removed",
+      transform: (value, context) =>
+        new Promise((resolve) => {
+          context.signal.addEventListener("abort", () => {
+            run.current.revision = 2
+            reentrant = coordinator.apply({ ...run.request, revision: 2 })
+            resolve(draft(`${value.text}:removed`))
+          })
+        }),
+    })
+    coordinator.register({ id: "survivor", transform: async (value) => draft(`${value.text}:survivor`) })
+    const active = coordinator.apply(run.request)
+    dispose()
+    expect(await Promise.all([active, reentrant])).toEqual([false, true])
+  })
+  test("commit reentrancy wins, exceptions recover, and stale branches reject", async () => {
+    const coordinator = createPromptTransformCoordinator<Draft, Context>(fingerprint)
+    coordinator.register({ id: "edit", transform: async (value) => draft(`${value.text}:done`) })
+    const committed = harness("commit", 1, "commit")
+    let nested!: Promise<boolean>
+    committed.request.commit = () => {
+      committed.current.revision = 2
+      nested = coordinator.apply({ ...committed.request, revision: 2, commit: () => undefined })
+      coordinator.disposeDraft(committed.request.draftID)
+      return undefined
+    }
+    expect(await coordinator.apply(committed.request)).toBe(true)
+    expect(await nested).toBe(false)
+    const run = harness("failure", 1, "failure")
+    run.request.commit = () => {
+      throw new Error("commit failed")
+    }
+    await expect(coordinator.apply(run.request)).rejects.toThrow("commit failed")
+    const next = draft("next")
+    expect(await coordinator.apply({ ...revise(run, next, 2), commit: () => undefined })).toBe(true)
+    const stale = draft("stale")
+    expect(await coordinator.apply(revise(run, stale, 1))).toBe(false)
+    const conflict = draft("conflict")
+    expect(await coordinator.apply(revise(run, conflict, 2))).toBe(false)
+  })
+})