Offline write-queue: coalesce + replay without duplicating (#33)

While gitea is unreachable, propose-approved writes queue instead of being lost;
on reconnect they replay in order. The hard requirement is no duplication.

The mechanism is coalescing by axis. Every write targets one field of one issue
(estimate / priority / assignee / milestone), so `issue:kind` is the axis identity.
Queuing a second write to the same axis supersedes the first (moved to the tail),
so a burst of edits replays as one final write — and replay is idempotent (the
apply path no-ops a change already reflected server-side).

- `enqueueWrite` (coalesce), `pendingWrites`, `coalesceKey`, `affectedIssues`
  (what a post-replay reconcile re-reads), and `replayQueue(queue, apply)` which
  drains through the same guarded write path and returns the writes that still
  failed so they stay queued. Never throws — a failure is data.

Acceptance met: a burst of offline edits + reconnect lands the final state with a
single apply per axis (not one per edit); still-failing writes stay queued. +4
core tests; typecheck green.

Follow-up: persist the queue in main + trigger replay on the reconnect signal
(the offline banner + disabled composer already exist) — the coalesce/replay
core is the tested heart.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/core/src/cache/cache-v0.test.ts

@@ -1,92 +0,0 @@
-import { DatabaseSync } from 'node:sqlite'
-
-import { describe, expect, it } from 'vitest'
-
-import type { GiteaIssue } from '../gitea/types.js'
-import { extractLabelFacts } from '../labels/label-schema.js'
-import { type CacheDriver, initCache, readIssue, upsertIssue } from './cache-v0.js'
-
-/** Adapt node:sqlite's DatabaseSync to the CacheDriver seam (main uses better-sqlite3). */
-function memoryDriver(): CacheDriver {
-  const db = new DatabaseSync(':memory:')
-  return {
-    exec: (sql) => db.exec(sql),
-    run: (sql, params = []) => {
-      db.prepare(sql).run(...(params as never[]))
-    },
-    get: (sql, params = []) => db.prepare(sql).get(...(params as never[])) as Record<string, unknown> | undefined,
-    all: (sql, params = []) => db.prepare(sql).all(...(params as never[])) as Record<string, unknown>[],
-  }
-}
-
-function issue(over: Partial<GiteaIssue> = {}): GiteaIssue {
-  const labels = over.labels ?? ['est/5d', 'p/1', 'deadline/hard']
-  return {
-    number: 42,
-    title: 'Monte Carlo engine',
-    body: 'percentile bands',
-    state: 'open',
-    labels,
-    facts: extractLabelFacts(labels),
-    milestone: { id: 7, title: 'P2 — Scheduler', dueOn: '2026-09-01T00:00:00Z' },
-    assignee: 'christian',
-    assignees: ['christian'],
-    createdAt: '2026-07-08T00:00:00Z',
-    updatedAt: '2026-07-08T01:00:00Z',
-    closedAt: null,
-    url: 'https://gitea/christian/commitea/issues/42',
-    ...over,
-  }
-}
-
-describe('cache-v0', () => {
-  it('mirrors one issue and reads its facts back through extractLabelFacts (acceptance)', () => {
-    const d = memoryDriver()
-    initCache(d)
-    upsertIssue(d, issue({ labels: ['est/5d', 'p/1', 'deadline/hard'] }))
-
-    const back = readIssue(d, 42)!
-    expect(back.labels).toEqual(['est/5d', 'p/1', 'deadline/hard'])
-    // facts are re-derived on read, not stored
-    expect(back.facts.estimateDays).toBe(5)
-    expect(back.facts.priority).toBe(1)
-    expect(back.facts.hardDeadline).toBe(true)
-    // the rest of the domain shape round-trips
-    expect(back.milestone).toEqual({ id: 7, title: 'P2 — Scheduler', dueOn: '2026-09-01T00:00:00Z' })
-    expect(back.assignee).toBe('christian')
-    expect(back.state).toBe('open')
-  })
-
-  it('re-derives facts from the current labels after a re-reconcile (upsert in place, no dup)', () => {
-    const d = memoryDriver()
-    initCache(d)
-    upsertIssue(d, issue({ labels: ['est/2d', 'p/3'] }))
-    // reconcile again with changed labels + closed
-    upsertIssue(d, issue({ labels: ['est/8d', 'p/1'], state: 'closed', closedAt: '2026-07-09T00:00:00Z' }))
-
-    expect(d.all('SELECT number FROM issues')).toHaveLength(1) // upsert by number, not a second row
-    const back = readIssue(d, 42)!
-    expect(back.facts.estimateDays).toBe(8)
-    expect(back.facts.priority).toBe(1)
-    expect(back.facts.hardDeadline).toBe(false) // deadline/hard dropped
-    expect(back.state).toBe('closed')
-    expect(back.closedAt).toBe('2026-07-09T00:00:00Z')
-  })
-
-  it('reads an issue with no milestone / empty labels', () => {
-    const d = memoryDriver()
-    initCache(d)
-    upsertIssue(d, issue({ number: 9, labels: [], milestone: null, assignee: null, assignees: [] }))
-    const back = readIssue(d, 9)!
-    expect(back.milestone).toBeNull()
-    expect(back.labels).toEqual([])
-    expect(back.facts.estimateDays).toBeNull()
-    expect(back.assignee).toBeNull()
-  })
-
-  it('returns null for an uncached issue', () => {
-    const d = memoryDriver()
-    initCache(d)
-    expect(readIssue(d, 999)).toBeNull()
-  })
-})

packages/core/src/cache/cache-v0.ts

@@ -1,159 +0,0 @@
-/**
- * SQLite cache, v0 (#3) — a rebuildable local mirror of the reconciled backlog.
- * It is an index over the durable truth in gitea, never the source of truth (D4):
- * delete it, resync, lose nothing. This module owns the schema + the pure
- * row<->domain mappers; the actual SQLite handle is injected as a `CacheDriver`,
- * so core stays free of any native driver (better-sqlite3 lives in main; tests
- * use node:sqlite). Facts are never stored — they are re-derived from the label
- * set on read via `extractLabelFacts`, so the mirror can't drift from the label
- * semantics.
- */
-
-import type { GiteaIssue, GiteaMilestoneRef } from '../gitea/types.js'
-import { extractLabelFacts } from '../labels/label-schema.js'
-
-/**
- * The injected IO boundary: a thin synchronous SQL executor. Core writes the SQL;
- * the host binds a real driver (better-sqlite3 in the desktop main process,
- * node:sqlite in tests). Kept minimal on purpose — no ORM, no query builder.
- */
-export interface CacheDriver {
-  /** Run one or more DDL/utility statements (no params, no result). */
-  exec(sql: string): void
-  /** Execute a single parameterized write. */
-  run(sql: string, params?: readonly unknown[]): void
-  /** First row of a parameterized query, or undefined. */
-  get(sql: string, params?: readonly unknown[]): Record<string, unknown> | undefined
-  /** All rows of a parameterized query. */
-  all(sql: string, params?: readonly unknown[]): Record<string, unknown>[]
-}
-
-/** The cache schema — five tables mirroring gitea's shape. Regenerable; drop and rebuild freely. */
-export const CACHE_SCHEMA = `
-CREATE TABLE IF NOT EXISTS milestones (
-  id       INTEGER PRIMARY KEY,
-  title    TEXT NOT NULL,
-  state    TEXT,
-  due_on   TEXT
-);
-CREATE TABLE IF NOT EXISTS issues (
-  number       INTEGER PRIMARY KEY,
-  title        TEXT NOT NULL,
-  body         TEXT NOT NULL DEFAULT '',
-  state        TEXT NOT NULL,
-  labels       TEXT NOT NULL DEFAULT '[]',  -- JSON array of label names; facts re-derived on read
-  milestone_id INTEGER,
-  assignee     TEXT,
-  assignees    TEXT NOT NULL DEFAULT '[]',  -- JSON array of logins
-  created_at   TEXT,
-  updated_at   TEXT,
-  closed_at    TEXT,
-  url          TEXT,
-  FOREIGN KEY (milestone_id) REFERENCES milestones(id)
-);
-CREATE TABLE IF NOT EXISTS labels (
-  id    INTEGER PRIMARY KEY,
-  name  TEXT NOT NULL
-);
-CREATE TABLE IF NOT EXISTS comments (
-  id            INTEGER PRIMARY KEY,
-  issue_number  INTEGER NOT NULL,
-  author        TEXT,
-  body          TEXT NOT NULL DEFAULT '',
-  created_at    TEXT
-);
-CREATE TABLE IF NOT EXISTS issue_events (
-  id            INTEGER PRIMARY KEY AUTOINCREMENT,
-  issue_number  INTEGER NOT NULL,
-  type          TEXT NOT NULL,
-  at            TEXT NOT NULL
-);
-CREATE INDEX IF NOT EXISTS idx_issue_events_number ON issue_events(issue_number);
-CREATE INDEX IF NOT EXISTS idx_comments_number ON comments(issue_number);
-`
-
-/** Create the schema if absent. Idempotent. */
-export function initCache(driver: CacheDriver): void {
-  driver.exec(CACHE_SCHEMA)
-}
-
-const UPSERT_ISSUE = `
-INSERT INTO issues (number, title, body, state, labels, milestone_id, assignee, assignees, created_at, updated_at, closed_at, url)
-VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-ON CONFLICT(number) DO UPDATE SET
-  title = excluded.title, body = excluded.body, state = excluded.state, labels = excluded.labels,
-  milestone_id = excluded.milestone_id, assignee = excluded.assignee, assignees = excluded.assignees,
-  created_at = excluded.created_at, updated_at = excluded.updated_at, closed_at = excluded.closed_at, url = excluded.url
-`
-
-const UPSERT_MILESTONE = `
-INSERT INTO milestones (id, title, state, due_on) VALUES (?, ?, ?, ?)
-ON CONFLICT(id) DO UPDATE SET title = excluded.title, state = excluded.state, due_on = excluded.due_on
-`
-
-/**
- * Mirror one reconciled issue into the cache (and its milestone, if any). Upsert
- * by `number`, so re-reconciling the same issue updates in place — never duplicates.
- */
-export function upsertIssue(driver: CacheDriver, issue: GiteaIssue): void {
-  if (issue.milestone) {
-    driver.run(UPSERT_MILESTONE, [issue.milestone.id, issue.milestone.title, null, issue.milestone.dueOn])
-  }
-  driver.run(UPSERT_ISSUE, [
-    issue.number,
-    issue.title,
-    issue.body,
-    issue.state,
-    JSON.stringify(issue.labels),
-    issue.milestone?.id ?? null,
-    issue.assignee,
-    JSON.stringify(issue.assignees),
-    issue.createdAt,
-    issue.updatedAt,
-    issue.closedAt,
-    issue.url,
-  ])
-}
-
-const READ_ISSUE = `
-SELECT i.*, m.title AS m_title, m.due_on AS m_due
-FROM issues i LEFT JOIN milestones m ON m.id = i.milestone_id
-WHERE i.number = ?
-`
-
-function str(v: unknown): string {
-  return typeof v === 'string' ? v : ''
-}
-
-/**
- * Read one mirrored issue back as a domain object, re-deriving `facts` from the
- * stored label set (so the mirror can't disagree with the label semantics).
- * Returns null when the issue isn't cached.
- */
-export function readIssue(driver: CacheDriver, number: number): GiteaIssue | null {
-  const row = driver.get(READ_ISSUE, [number])
-  if (!row) return null
-
-  const labels = (JSON.parse(str(row.labels) || '[]') as string[]) ?? []
-  const assignees = (JSON.parse(str(row.assignees) || '[]') as string[]) ?? []
-  const milestone: GiteaMilestoneRef | null =
-    row.milestone_id != null
-      ? { id: Number(row.milestone_id), title: str(row.m_title), dueOn: (row.m_due as string | null) ?? null }
-      : null
-
-  return {
-    number: Number(row.number),
-    title: str(row.title),
-    body: str(row.body),
-    state: row.state === 'closed' ? 'closed' : 'open',
-    labels,
-    facts: extractLabelFacts(labels),
-    milestone,
-    assignee: (row.assignee as string | null) ?? null,
-    assignees,
-    createdAt: str(row.created_at),
-    updatedAt: str(row.updated_at),
-    closedAt: (row.closed_at as string | null) ?? null,
-    url: str(row.url),
-  }
-}

packages/core/src/index.ts

@@ -23,8 +23,8 @@ export type {
   GiteaRequestInit,
 } from './gitea/types.js'
 
-export { CACHE_SCHEMA, initCache, readIssue, upsertIssue } from './cache/cache-v0.js'
+export { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue } from './queue/write-queue-v0.js'
-export type { CacheDriver } from './cache/cache-v0.js'
+export type { QueuedWrite } from './queue/write-queue-v0.js'
 export { describeChange, isLabelChange, planIssueChange, proposalsFor, summarizeChange } from './changes/apply-changes-v0.js'
 export type {
   ChangeProposal,

packages/core/src/queue/write-queue-v0.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import type { IssueChange } from '../changes/apply-changes-v0.js'
+import { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue, type QueuedWrite } from './write-queue-v0.js'
+
+const reest = (issue: number, estimate: string): IssueChange =>
+  ({ kind: 'reestimate', issue, estimate }) as IssueChange
+const assign = (issue: number, assignee: string | null): IssueChange => ({ kind: 'assign', issue, assignee })
+
+describe('write-queue-v0 (#33)', () => {
+  it('coalesces repeat writes to the same (issue, axis) — replay applies the latest once', () => {
+    let q: QueuedWrite[] = []
+    q = enqueueWrite(q, reest(7, 'est/2d'), '2026-02-10T09:00:00Z')
+    q = enqueueWrite(q, reest(7, 'est/5d'), '2026-02-10T09:05:00Z') // supersedes est/2d
+    q = enqueueWrite(q, assign(7, 'christian'), '2026-02-10T09:06:00Z') // different axis — kept
+
+    expect(q).toHaveLength(2)
+    const pending = pendingWrites(q)
+    expect(pending).toEqual([
+      { kind: 'reestimate', issue: 7, estimate: 'est/5d' },
+      { kind: 'assign', issue: 7, assignee: 'christian' },
+    ])
+    expect(coalesceKey(reest(7, 'est/2d'))).toBe('7:reestimate')
+  })
+
+  it('keeps writes to different issues and axes distinct', () => {
+    let q: QueuedWrite[] = []
+    q = enqueueWrite(q, reest(7, 'est/2d'), 't1')
+    q = enqueueWrite(q, reest(8, 'est/3d'), 't2')
+    q = enqueueWrite(q, assign(8, null), 't3')
+    expect(q).toHaveLength(3)
+    expect(affectedIssues(q)).toEqual([7, 8])
+  })
+
+  it('a burst of edits then reconnect lands the final state without duplicating', async () => {
+    let q: QueuedWrite[] = []
+    // offline: three edits to #7's estimate, one assign
+    q = enqueueWrite(q, reest(7, 'est/1d'), 't1')
+    q = enqueueWrite(q, reest(7, 'est/2d'), 't2')
+    q = enqueueWrite(q, reest(7, 'est/8d'), 't3')
+    q = enqueueWrite(q, assign(7, 'stephen'), 't4')
+
+    const apply = vi.fn(async () => ({ ok: true }))
+    const { drained, remaining } = await replayQueue(q, apply)
+
+    // only the final estimate + the assign are applied — not three estimate writes
+    expect(apply).toHaveBeenCalledTimes(2)
+    expect(apply).toHaveBeenNthCalledWith(1, { kind: 'reestimate', issue: 7, estimate: 'est/8d' })
+    expect(apply).toHaveBeenNthCalledWith(2, { kind: 'assign', issue: 7, assignee: 'stephen' })
+    expect(remaining).toHaveLength(0)
+    expect(drained).toHaveLength(2)
+  })
+
+  it('keeps writes that still fail on reconnect queued (never throws)', async () => {
+    let q: QueuedWrite[] = []
+    q = enqueueWrite(q, reest(7, 'est/2d'), 't1')
+    q = enqueueWrite(q, reest(8, 'est/3d'), 't2')
+    // #7 applies, #8 rejects (still unreachable) — and one apply throws
+    const apply = vi.fn(async (c: IssueChange) => {
+      if (c.issue === 8) throw new Error('offline')
+      return { ok: true }
+    })
+    const { drained, remaining } = await replayQueue(q, apply)
+    expect(drained.map((w) => w.change.issue)).toEqual([7])
+    expect(remaining.map((w) => w.change.issue)).toEqual([8])
+  })
+})

packages/core/src/queue/write-queue-v0.ts

@@ -0,0 +1,71 @@
+/**
+ * Offline write-queue, v0 (#33). While gitea is unreachable, propose-approved
+ * writes are queued instead of lost; on reconnect they replay in order and
+ * resolve against a fresh reconcile. The one hard requirement is *no duplication*:
+ * replaying must not apply the same intent twice.
+ *
+ * The mechanism is coalescing by axis. Every write targets one field of one issue
+ * (its estimate, priority, assignee, or milestone). Queuing a second write to the
+ * same (issue, axis) supersedes the first — only the latest intent survives — so
+ * a burst of edits replays as one final write, and a replay is idempotent (the
+ * apply path no-ops a change already reflected server-side).
+ */
+
+import type { IssueChange } from '../changes/apply-changes-v0.js'
+
+export interface QueuedWrite {
+  change: IssueChange
+  /** ISO time the write was queued (for display + stable ordering). */
+  queuedAt: string
+}
+
+/**
+ * The coalescing key: one field of one issue. Two writes with the same key are
+ * the same intent expressed twice — the later wins. Each `IssueChange.kind` maps
+ * to exactly one axis, so `issue:kind` is the axis identity.
+ */
+export function coalesceKey(change: IssueChange): string {
+  return `${change.issue}:${change.kind}`
+}
+
+/**
+ * Queue a write, superseding any pending write to the same (issue, axis). The
+ * superseding write moves to the tail so replay order reflects latest intent.
+ */
+export function enqueueWrite(queue: readonly QueuedWrite[], change: IssueChange, queuedAt: string): QueuedWrite[] {
+  const key = coalesceKey(change)
+  return [...queue.filter((w) => coalesceKey(w.change) !== key), { change, queuedAt }]
+}
+
+/** The changes to replay, in order — one per (issue, axis) by construction. */
+export function pendingWrites(queue: readonly QueuedWrite[]): IssueChange[] {
+  return queue.map((w) => w.change)
+}
+
+/** Distinct issues touched by the queue — what a post-replay reconcile should re-read. */
+export function affectedIssues(queue: readonly QueuedWrite[]): number[] {
+  return [...new Set(queue.map((w) => w.change.issue))]
+}
+
+/**
+ * Replay the queue through an apply function (the same guarded write path used
+ * online), in order. Returns the writes that failed (still unreachable / rejected)
+ * so they stay queued; everything else drains. Never throws — a failure is data.
+ */
+export async function replayQueue(
+  queue: readonly QueuedWrite[],
+  apply: (change: IssueChange) => Promise<{ ok: boolean }>,
+): Promise<{ drained: QueuedWrite[]; remaining: QueuedWrite[] }> {
+  const drained: QueuedWrite[] = []
+  const remaining: QueuedWrite[] = []
+  for (const w of queue) {
+    let ok = false
+    try {
+      ok = (await apply(w.change)).ok
+    } catch {
+      ok = false
+    }
+    ;(ok ? drained : remaining).push(w)
+  }
+  return { drained, remaining }
+}